fix: refactor code for enrollment api to make code cleaner

futurex_openedx_extensions/dashboard/details/learners.py

@@ -11,13 +11,11 @@
 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,
+    get_course_search_queryset,
     get_learners_search_queryset,
     get_one_user_queryset,
     get_permitted_learners_queryset,
@@ -284,36 +282,20 @@ def get_learners_enrollments_queryset(  # pylint: disable=too-many-arguments
     :return: List of dictionaries containing user and course details.
     """
     accessible_users = get_permitted_learners_queryset(
-        queryset=get_learners_search_queryset(search_text),
+        queryset=get_learners_search_queryset(
+            search_text=learner_search,
+            user_ids=user_ids,
+            usernames=usernames,
+        ),
         fx_permission_info=fx_permission_info,
         include_staff=include_staff,
     )
-    user_filter = Q()
-    if user_ids:
-        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}',
-            )
-        user_filter |= Q(id__in=user_ids)
-    if usernames:
-        invalid_usernames = [username for username in usernames if not isinstance(username, str)]
-        if invalid_usernames:
-            raise FXCodedException(
-                code=FXExceptionCodes.INVALID_INPUT,
-                message=f'Invalid usernames: {invalid_usernames}',
-            )
-        user_filter |= Q(username__in=usernames)
-    accessible_users = accessible_users.filter(user_filter)
 
-    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'])
+    accessible_courses = get_course_search_queryset(
+        fx_permission_info=fx_permission_info,
+        search_text=course_search,
+        course_ids=course_ids,
     )
-    if course_ids:
-        verify_course_ids(course_ids)
-        accessible_courses = accessible_courses.filter(id__in=course_ids)
 
     queryset = CourseEnrollment.objects.filter(
         is_active=True,

futurex_openedx_extensions/helpers/querysets.py

@@ -12,7 +12,7 @@
 
 from futurex_openedx_extensions.helpers.converters import get_allowed_roles
 from futurex_openedx_extensions.helpers.exceptions import FXCodedException, FXExceptionCodes
-from futurex_openedx_extensions.helpers.extractors import get_partial_access_course_ids
+from futurex_openedx_extensions.helpers.extractors import get_partial_access_course_ids, verify_course_ids
 from futurex_openedx_extensions.helpers.tenants import get_tenants_sites
 from futurex_openedx_extensions.helpers.users import get_user_by_key
 from futurex_openedx_extensions.upgrade.models_switch import CourseAccessRole, CourseEnrollment, UserSignupSource
@@ -208,11 +208,13 @@ def get_base_queryset_courses(
     return q_set
 
 
-def get_learners_search_queryset(
+def get_learners_search_queryset(  # pylint: disable=too-many-arguments
     search_text: str | None = None,
     superuser_filter: bool | None = False,
     staff_filter: bool | None = False,
-    active_filter: bool | None = True
+    active_filter: bool | None = True,
+    user_ids: List[int] | None = None,
+    usernames: List[str] | None = None,
 ) -> QuerySet:
     """
     Get the learners queryset for the given search text.
@@ -246,6 +248,59 @@ def get_learners_search_queryset(
             Q(profile__name__icontains=search_text)
         )
 
+    user_filter = Q()
+    if user_ids:
+        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}',
+            )
+        user_filter |= Q(id__in=user_ids)
+    if usernames:
+        invalid_usernames = [username for username in usernames if not isinstance(username, str)]
+        if invalid_usernames:
+            raise FXCodedException(
+                code=FXExceptionCodes.INVALID_INPUT,
+                message=f'Invalid usernames: {invalid_usernames}',
+            )
+        user_filter |= Q(username__in=usernames)
+
+    queryset = queryset.filter(user_filter)
+
+    return queryset
+
+
+def get_course_search_queryset(
+    fx_permission_info: dict,
+    search_text: str | None = None,
+    course_ids: List[str] | None = None,
+) -> QuerySet:
+    """
+    Get the courses queryset for the given search text.
+
+    :param fx_permission_info: Dictionary containing permission information
+    :type fx_permission_info: dict
+    :param search_text: Search text to filter the courses by
+    :type search_text: str | None
+    :return: QuerySet of courses
+    :rtype: QuerySet
+    """
+    queryset = 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)
+        queryset = queryset.filter(id__in=course_ids)
+
+    course_search = (search_text or '').strip()
+    if course_search:
+        queryset = queryset.filter(
+            Q(display_name__icontains=course_search) |
+            Q(id__icontains=course_search),
+        )
+
     return queryset
 
 

tests/test_dashboard/test_details/test_details_learners.py

@@ -224,35 +224,53 @@ def test_get_learners_enrollments_queryset_annotations(
     ).count() == 0, 'only active enrollments should be filtered'
 
 
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    'course_search, learner_search, expected_count, usecase',
+    [
+        ('', '', 25, 'no search'),
+        ('Course 5', '', 8, 'only course search'),
+        ('', 'user15', 2, 'only user search'),
+        ('Course 5', 'user15', 1, 'both course and user search'),
+    ],
+)
+def test_get_learners_enrollments_queryset_for_course_and_learner_search(
+    course_search, learner_search, expected_count, usecase, fx_permission_info,
+):
+    """Test get_learners_by_course_queryset result for accessible users and courses."""
+    fx_permission_info['view_allowed_full_access_orgs'] = ['org1', 'org2']
+    queryset = get_learners_enrollments_queryset(
+        fx_permission_info=fx_permission_info,
+        course_search=course_search,
+        learner_search=learner_search
+    )
+    assert queryset.count() == expected_count, f'unexpected enrollment queryset count for case: {usecase}'
+
+
 @pytest.mark.django_db
 @pytest.mark.parametrize(
     'filter_user_ids, filter_course_ids, filter_usernames, expected_error, expected_count, usecase', [
-        (None, None, None, '', 6, 'No filter'),
+        (None, None, None, '', 25, 'No filter'),
         ([21], None, None, '', 3, 'only user_ids filter'),
         (None, None, ['user21'], '', 3, 'only usernames filter'),
         ([21], None, ['user29'], '', 4, 'user_ids and usernames filter'),
-        ([], ['course-v1:ORG1+5+5'], [], '', 2, 'only course_ids filter'),
+        ([], ['course-v1:ORG1+5+5'], [], '', 3, 'only course_ids filter'),
         ([15, 29], ['course-v1:ORG1+5+5'], [], '', 1, 'both course_ids and user_ids filter'),
         ([15], ['course-v1:ORG1+5+5'], ['user21'], '', 2, 'course_ids, usernames and user_ids filter'),
+        (
+            [15], ['course-v1:ORG1+5+5'], ['user21', 'user15'], '', 2,
+            'course_ids, usernames and user_ids filter, with repeated user in user ids and usernames'
+        ),
         ([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, 'invalid course_ids'),
         ([], [], [11], 'Invalid usernames: [11]', 0, 'invalid usernames'),
     ]
 )
-@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, filter_usernames,
-    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)
-
+def test_get_learners_enrollments_queryset_for_course_and_user_filters(
+    filter_user_ids, filter_course_ids, filter_usernames, expected_error, expected_count, usecase, fx_permission_info,
+):  # pylint: disable=too-many-arguments
+    """Test get_learners_by_course_queryset result for user and course filters."""
+    fx_permission_info['view_allowed_full_access_orgs'] = ['org1', 'org2']
     if expected_error:
         with pytest.raises(FXCodedException) as exc_info:
             queryset = get_learners_enrollments_queryset(
@@ -269,68 +287,4 @@ def test_get_learners_enrollments_queryset_for_course_ids_and_user_ids_filter(
             user_ids=filter_user_ids,
             usernames=filter_usernames,
         )
-        expected_user_ids = filter_user_ids or fake_accessible_user_ids
-        if filter_usernames:
-            expected_user_ids.extend(
-                get_user_model().objects.filter(
-                    username__in=filter_usernames
-                ).values_list('id', flat=True)
-            )
-        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_helpers/test_querysets.py

@@ -179,6 +179,90 @@ def test_get_learners_search_queryset_active_filter(
     assert querysets.get_learners_search_queryset(**kwargs).count() == 70 - true_count
 
 
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    'user_ids_filter, usernames_filter, expected_count, expected_error, usecase',
+    [
+        ([], [], 64, None, 'no filter'),
+        ([15, 21], [], 2, None, 'valid, only user ids filter for existing users'),
+        ([15, 100000001], [], 1, None, 'valid, user ids filter with non-existing users'),
+        (None, ['user15', 'user29'], 2, None, 'valid, only usernames filter with existing usernames'),
+        (None, ['user15', 'non-exist'], 1, None, 'valid, only usernames filter with non-existing username'),
+        ([21], ['user15'], 2, None, 'valid, both filters'),
+        ([21, 15], ['user15'], 2, None, 'valid, both filters with repeated user'),
+        ([15, 29, 'not-int', 0.2], None, 0, 'Invalid user ids: [\'not-int\', 0.2]', 'invalid user ids'),
+        (None, [11], 0, 'Invalid usernames: [11]', 'invalid usernames'),
+    ]
+)
+def test_get_learners_search_queryset_for_userids_and_usernames(
+    base_data, user_ids_filter, usernames_filter, expected_count, expected_error, usecase
+):  # pylint: disable=too-many-arguments, unused-argument
+    """
+    Verify that get_learners_search_queryset returns the correct QuerySet when userids and usernames filters are used
+    """
+    if expected_error:
+        with pytest.raises(FXCodedException) as exc_info:
+            queryset = querysets.get_learners_search_queryset(user_ids=user_ids_filter, usernames=usernames_filter)
+        assert str(exc_info.value) == expected_error
+    else:
+        queryset = querysets.get_learners_search_queryset(user_ids=user_ids_filter, usernames=usernames_filter)
+        assert queryset.count() == expected_count, f'unexpected learners queryset count for case: {usecase}'
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    'limited_access_course_ids, full_access_orgs, expected_count, usecase',
+    [
+        ([], ['org1', 'org2'], 12, 'full access orgs'),
+        (['course-v1:ORG1+5+5'], [], 1, 'course limited access'),
+        (['course-v1:ORG2+4+4'], ['org1'], 6, 'full access orgs with course limited access'),
+    ],
+)
+@patch('futurex_openedx_extensions.helpers.querysets.get_partial_access_course_ids')
+def test_get_course_search_queryset(
+    mocked_partial_course_ids, limited_access_course_ids,
+    full_access_orgs, expected_count, usecase, fx_permission_info,
+):  # pylint: disable=too-many-arguments
+    """Test get_course_search_queryset result"""
+    mocked_partial_course_ids.return_value = limited_access_course_ids
+    fx_permission_info['view_allowed_full_access_orgs'] = full_access_orgs
+    queryset = querysets.get_course_search_queryset(
+        fx_permission_info=fx_permission_info,
+    )
+    assert queryset.count() == expected_count, f'unexpected courses queryset count for case: {usecase}'
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    'search_text, course_ids_filter, expected_count, expected_error, usecase',
+    [
+        (None, None, 12, None, 'valid: no filter or search'),
+        ('Course 5', None, 2, None, 'valid: search with matching course'),
+        ('non-exist', None, 0, None, 'valid: search with no matching course'),
+        ('', ['course-v1:ORG2+4+4'], 1, None, 'valid: filter by course ID'),
+        ('', ['course-v1:ORG2+4+4', 'invalid'], 0, 'Invalid course ID format: invalid', 'invalid: course ID format'),
+        ('', ['course-v1:ORG2+4+4', 'course-v1:ORG1+5+1000'], 1, None, 'valid: filter with one non-existent course ID'),
+        ('course 4', ['course-v1:ORG2+4+4'], 1, None, 'valid: search and course id filter'),
+        ('course 5', ['course-v1:ORG2+4+4'], 0, None, 'valid: no course matches the condition'),
+    ],
+)
+def test_get_course_search_queryset_for_search_and_filter(
+    search_text, course_ids_filter, expected_count, expected_error, usecase, fx_permission_info,
+):  # pylint: disable=too-many-arguments
+    """Test get_course_search_queryset result for search and course ids filter"""
+    fx_permission_info['view_allowed_full_access_orgs'] = ['org1', 'org2']
+    if expected_error:
+        with pytest.raises(FXCodedException) as exc_info:
+            querysets.get_course_search_queryset(
+                fx_permission_info, search_text=search_text, course_ids=course_ids_filter
+            )
+        assert str(exc_info.value) == expected_error
+    else:
+        assert querysets.get_course_search_queryset(
+            fx_permission_info, search_text=search_text, course_ids=course_ids_filter
+        ).count() == expected_count, f'unexpected courses queryset count for case: {usecase}'
+
+
 @pytest.mark.django_db
 @pytest.mark.parametrize('full_access, partial_access, expected_with_staff, expected_without_staff', [
     ([7, 8], [], 26, 22),