Fix broken tests

futurex_openedx_extensions/dashboard/details/learners.py

@@ -3,11 +3,11 @@
 
 from datetime import timedelta
 
+from common.djangoapps.student.models import CourseEnrollment
 from django.contrib.auth import get_user_model
 from django.db.models import BooleanField, Case, Count, Exists, OuterRef, Q, Subquery, Value, When
 from django.db.models.query import QuerySet
 from django.utils import timezone
-from common.djangoapps.student.models import CourseEnrollment
 from lms.djangoapps.certificates.models import GeneratedCertificate
 from lms.djangoapps.courseware.models import StudentModule
 from lms.djangoapps.grades.models import PersistentCourseGrade
@@ -253,9 +253,11 @@ def get_learner_info_queryset(
     return queryset
 
 
-def get_learners_enrollments_queryset(user_ids=None, course_ids=None):
+def get_learners_enrollments_queryset(
+    user_ids: list = None, course_ids: list = None, include_staff: bool = False
+) -> QuerySet:
     """
-    Get the learners and their course-specific details. If no course_ids or user_ids are provided, 
+    Get the enrollment details. If no course_ids or user_ids are provided,
     all relevant data will be processed.
 
     :param course_ids: List of course IDs to filter by (optional).
@@ -265,16 +267,19 @@ def get_learners_enrollments_queryset(user_ids=None, course_ids=None):
     :return: List of dictionaries containing user and course details.
     """
 
-    course_filter = Q(is_active=True) 
-    if course_ids and user_ids:
-        course_filter &= Q(course_id__in=course_ids, user_id__in=user_ids)
-    elif course_ids:
+    course_filter = Q(is_active=True)
+    if course_ids:
         course_filter &= Q(course_id__in=course_ids)
-    elif user_ids:
+    if user_ids:
         course_filter &= Q(user_id__in=user_ids)
-    
+
     queryset = CourseEnrollment.objects.filter(course_filter)
-
+
+    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'),
+        )
+
     queryset = queryset.annotate(
         certificate_available=Exists(
             GeneratedCertificate.objects.filter(

futurex_openedx_extensions/dashboard/serializers.py

@@ -4,13 +4,13 @@
 import re
 from typing import Any, Dict, Tuple
 
+from common.djangoapps.student.models import CourseEnrollment
 from django.contrib.auth import get_user_model
 from django.utils.timezone import now
 from lms.djangoapps.courseware.courses import get_course_blocks_completion_summary
 from lms.djangoapps.grades.api import CourseGradeFactory
 from lms.djangoapps.grades.context import grading_context_for_course
 from lms.djangoapps.grades.models import PersistentSubsectionGrade
-from common.djangoapps.student.models import CourseEnrollment
 from opaque_keys.edx.locator import CourseLocator
 from openedx.core.djangoapps.content.block_structure.api import get_block_structure_manager
 from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
@@ -256,23 +256,21 @@ def __init__(self, *args: Any, **kwargs: Any):
         """Initialize the serializer."""
         super().__init__(*args, **kwargs)
         if not hasattr(self, '_get_user'):
-            raise NotImplementedError("Child class must implement '_get_user' method.")
-        
+            raise NotImplementedError('Child class must implement _get_user method.')
+
         if not hasattr(self, '_get_course_id'):
-            raise NotImplementedError("Child class must implement '_get_course_id' method.")
+            raise NotImplementedError('Child class must implement _get_course_id method.')
 
         self._is_exam_name_in_header = self.context.get('omit_subsection_name', '0') != '1'
         self._grading_info: Dict[str, Any] = {}
         self._subsection_locations: Dict[str, Any] = {}
 
-    def collect_grading_info(self, course_ids) -> None:
+    def collect_grading_info(self, course_ids: list) -> None:
         """Collect the grading info."""
         self._grading_info = {}
         self._subsection_locations = {}
-
         if not self.is_optional_field_requested('exam_scores'):
             return
-
         for course_id in course_ids:
             inc = 0
             grading_context = grading_context_for_course(get_course_by_id(course_id))
@@ -283,7 +281,7 @@ def collect_grading_info(self, course_ids) -> None:
                     if self.is_exam_name_in_header:
                         header_name += f': {subsection_info["subsection_block"].display_name}'
 
-                    index = f'{course_id}__{inc}'
+                    index = f'{course_id}_{inc}'
                     self._grading_info[str(index)] = {
                         'header_name': header_name,
                         'location': str(subsection_info['subsection_block'].location),
@@ -305,22 +303,26 @@ def grading_info(self) -> Dict[str, Any]:
     def subsection_locations(self) -> Dict[str, Any]:
         """Get the subsection locations."""
         return self._subsection_locations
-        
+
     def get_certificate_url(self, obj: Any) -> Any:
         """Return the certificate URL."""
-        return get_certificate_url(self.context.get('request'), self._get_user(obj), self._get_course_id(obj))
+        return get_certificate_url(
+            self.context.get('request'), self._get_user(obj), self._get_course_id(obj)   # pylint: disable=no-member
+        )
 
     def get_progress(self, obj: CourseEnrollment) -> Any:
         """Return the certificate URL."""
-        progress_info = get_course_blocks_completion_summary( self._get_course_id(obj), self._get_user(obj))
+        progress_info = get_course_blocks_completion_summary(
+            self._get_course_id(obj), self._get_user(obj)  # pylint: disable=no-member
+        )
         total = progress_info['complete_count'] + progress_info['incomplete_count'] + progress_info['locked_count']
         return round(progress_info['complete_count'] / total, 4) if total else 0.0
 
     def get_exam_scores(self, obj: get_user_model) -> Dict[str, Tuple[float, float] | None]:
         """Return exam scores."""
         result: Dict[str, Tuple[float, float] | None] = {__index: None for __index in self.grading_info}
         grades = PersistentSubsectionGrade.objects.filter(
-            user_id=self._get_user(obj).id,
+            user_id=self._get_user(obj).id,  # pylint: disable=no-member
             course_id=self._get_course_id(obj),
             usage_key__in=self.subsection_locations.keys(),
             first_attempted__isnull=False,
@@ -344,7 +346,9 @@ def _extract_exam_scores(representation_item: dict[str, Any]) -> None:
         return representation
 
 
-class LearnerDetailsForCourseSerializer(LearnerBasicDetailsSerializer, CourseScoreAndCertificateSerializer):
+class LearnerDetailsForCourseSerializer(
+    LearnerBasicDetailsSerializer, CourseScoreAndCertificateSerializer
+):  # pylint: disable=too-many-ancestors
     """Serializer for learner details for a course."""
 
     class Meta:
@@ -357,11 +361,11 @@ def __init__(self, *args: Any, **kwargs: Any):
         self._course_id = CourseLocator.from_string(self.context.get('course_id'))
         self.collect_grading_info([self._course_id])
 
-    def _get_course_id(self, obj: Any = None) -> CourseLocator:
+    def _get_course_id(self, obj: Any = None) -> CourseLocator:   # pylint: disable=unused-argument
         """Get the course ID. Its helper method required for CourseScoreAndCertificateSerializer"""
         return self._course_id
 
-    def _get_user(self, obj: Any = None) -> get_user_model():
+    def _get_user(self, obj: Any = None) -> get_user_model:   # pylint: disable=no-self-use
         """Get the User. Its helper method required for CourseScoreAndCertificateSerializer"""
         return obj
 
@@ -380,20 +384,20 @@ def __init__(self, *args: Any, **kwargs: Any):
         course_ids = self.context.get('course_ids')
         self.collect_grading_info(course_ids)
 
-    def _get_course_id(self, obj: Any = None) -> CourseLocator:
+    def _get_course_id(self, obj: Any = None) -> CourseLocator | None:   # pylint: disable=no-self-use
         """Get the course ID. Its helper method required for CourseScoreAndCertificateSerializer"""
-        return obj.course_id
+        return obj.course_id if obj else None
 
-    def _get_user(self, obj: Any = None) -> get_user_model():
+    def _get_user(self, obj: Any = None) -> get_user_model | None:   # pylint: disable=no-self-use
         """Get the User. Its helper method required for CourseScoreAndCertificateSerializer"""
-        return obj.user
+        return obj.user if obj else None
 
-    def get_course_id(self, obj) -> str:
+    def get_course_id(self, obj: Any) -> str:
         """Get course id"""
         return str(self._get_course_id(obj))
 
-    def to_representation(self, instance):
-        representation = LearnerBasicDetailsSerializer(instance.user).data
+    def to_representation(self, instance: Any) -> Any:
+        representation = LearnerBasicDetailsSerializer(self._get_user(instance)).data
         representation.update(super().to_representation(instance))
         return representation
 

futurex_openedx_extensions/dashboard/urls.py

@@ -26,7 +26,7 @@
         views.LearnersDetailsForCourseView.as_view(), name='learners-course'),
     re_path(
         r'^api/fx/learners/v1/enrollments/$',
-        views.LearnersEnrollmentView.as_view(), name='learners-enrollement'),
+        views.LearnersEnrollmentView.as_view(), name='learners-enrollements'),
     re_path(
         r'^api/fx/learners/v1/learner/' + settings.USERNAME_PATTERN + '/$',
         views.LearnerInfoView.as_view(),

futurex_openedx_extensions/dashboard/views.py

@@ -25,8 +25,8 @@
 from futurex_openedx_extensions.dashboard.details.learners import (
     get_learner_info_queryset,
     get_learners_by_course_queryset,
+    get_learners_enrollments_queryset,
     get_learners_queryset,
-    get_learners_enrollments_queryset
 )
 from futurex_openedx_extensions.dashboard.statistics.certificates import get_certificates_count
 from futurex_openedx_extensions.dashboard.statistics.courses import (
@@ -437,7 +437,7 @@ class LearnersEnrollmentView(ListAPIView, FXViewRoleInfoMixin):
     serializer_class = serializers.LearnerEnrollmentSerializer
     permission_classes = [FXHasTenantCourseAccess]
     pagination_class = DefaultPagination
-    fx_view_name = 'learners_with_enrollment_details'
+    fx_view_name = 'learners_enrollment_details'
     fx_default_read_only_roles = ['staff', 'instructor', 'data_researcher', 'org_course_creator_group']
     fx_view_description = 'api/fx/learners/v1/enrollments: Get the list of enrollemts'
 
@@ -446,6 +446,7 @@ def get_queryset(self, *args: Any, **kwargs: Any) -> QuerySet:
         return get_learners_enrollments_queryset(
             user_ids=self.request.query_params.getlist('user_ids', None),
             course_ids=self.request.query_params.getlist('course_ids', None),
+            include_staff=self.request.query_params.get('include_staff', '0') == '1'
         )
 
     def get_serializer_context(self) -> Dict[str, Any]:

tests/test_dashboard/test_details/test_details_learners.py

@@ -11,6 +11,7 @@
     get_courses_count_for_learner_queryset,
     get_learner_info_queryset,
     get_learners_by_course_queryset,
+    get_learners_enrollments_queryset,
     get_learners_queryset,
 )
 from futurex_openedx_extensions.helpers.exceptions import FXCodedException, FXExceptionCodes
@@ -193,3 +194,37 @@ def test_get_learners_by_course_queryset_include_staff(base_data):  # pylint: di
 
     queryset = get_learners_by_course_queryset('course-v1:ORG1+5+5', include_staff=True)
     assert queryset.count() == 5, 'unexpected test data'
+
+
+@pytest.mark.django_db
+def test_get_learners_enrollments_queryset_annotations(base_data):  # 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])
+    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, \
+        'course_score should be in the queryset with value 0.67'
+    assert queryset[0].active_in_course is False, \
+        'active_in_course should be in the queryset with value True'
+
+    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'
+
+
+@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}'

tests/test_dashboard/test_serializers.py

@@ -3,7 +3,7 @@
 
 import pytest
 from cms.djangoapps.course_creators.models import CourseCreator
-from common.djangoapps.student.models import CourseAccessRole, SocialLink, UserProfile
+from common.djangoapps.student.models import CourseAccessRole, CourseEnrollment, SocialLink, UserProfile
 from custom_reg_form.models import ExtraInfo
 from deepdiff import DeepDiff
 from django.contrib.auth import get_user_model
@@ -17,12 +17,14 @@
 from futurex_openedx_extensions.dashboard.serializers import (
     CourseDetailsBaseSerializer,
     CourseDetailsSerializer,
+    CourseScoreAndCertificateSerializer,
     DataExportTaskSerializer,
     LearnerBasicDetailsSerializer,
     LearnerCoursesDetailsSerializer,
     LearnerDetailsExtendedSerializer,
     LearnerDetailsForCourseSerializer,
     LearnerDetailsSerializer,
+    LearnerEnrollmentSerializer,
     UserRolesSerializer,
 )
 from futurex_openedx_extensions.helpers import constants as cs
@@ -253,13 +255,39 @@ def test_learner_details_serializer(base_data):  # pylint: disable=unused-argume
 
 
 @pytest.mark.django_db
-@patch('futurex_openedx_extensions.dashboard.serializers.LearnerDetailsForCourseSerializer.collect_grading_info')
-def test_learner_details_for_course_serializer(mock_collect, base_data,):  # pylint: disable=unused-argument
+def test_course_score_and_certificate_serializer_for_required_child_methods():
+    """Verify that the CourseScoreAndCertificateSerializer for required child methods"""
+    class TestSerializer(CourseScoreAndCertificateSerializer):
+        """Serializer for learner details for a course."""
+        class Meta:
+            model = get_user_model()
+            fields = CourseScoreAndCertificateSerializer.Meta.fields
+    qs = get_dummy_queryset()
+    with pytest.raises(NotImplementedError) as exc_info:
+        TestSerializer(qs, many=True)
+    assert str(exc_info.value) == 'Child class must implement _get_user method.'
+    TestSerializer._get_user = Mock()  # pylint: disable=protected-access
+    with pytest.raises(NotImplementedError) as exc_info:
+        TestSerializer(qs, many=True)
+    assert str(exc_info.value) == 'Child class must implement _get_course_id method.'
+    TestSerializer._get_course_id = Mock()  # pylint: disable=protected-access
+    serializer = TestSerializer(qs, many=True)
+    assert len(serializer.data) == qs.count()
+
+
+@pytest.mark.django_db
+@patch('futurex_openedx_extensions.dashboard.serializers.CourseScoreAndCertificateSerializer.collect_grading_info')
+def test_learner_enrollments_serializer(mock_collect, base_data,):  # pylint: disable=unused-argument
     """Verify that the LearnerDetailsForCourseSerializer returns the needed fields."""
-    queryset = get_dummy_queryset()
-    serializer = LearnerDetailsForCourseSerializer(queryset, context={'course_id': 'course-v1:ORG2+1+1'}, many=True)
+    queryset = CourseEnrollment.objects.filter(user_id=10, course_id='course-v1:ORG3+1+1').annotate(
+        certificate_available=Value(True),
+        course_score=Value(0.67),
+        active_in_course=Value(True),
+    )
+    serializer = LearnerEnrollmentSerializer(queryset, context={
+        'course_ids': ['course-v1:ORG3+1+1']
+    }, many=True)
     mock_collect.assert_called_once()
-
     data = serializer.data
     assert len(data) == 1
     assert data[0]['certificate_available'] is True
@@ -296,28 +324,27 @@ def test_learner_details_for_course_serializer_collect_grading_info(
     )
 
     expected_grading_info = {
-        '0': {
+        'course-v1:ORG2+1+1_0': {
             'header_name': 'Homework 1: First Homework' if display_name_included else 'Homework 1',
             'location': 'block-v1:ORG2+1+1+type@homework+block@1',
         },
-        '1': {
+        'course-v1:ORG2+1+1_1': {
             'header_name': 'Homework 2: Second Homework' if display_name_included else 'Homework 2',
             'location': 'block-v1:ORG2+1+1+type@homework+block@2',
         },
-        '2': {
+        'course-v1:ORG2+1+1_2': {
             'header_name': 'Exam: The final exam' if display_name_included else 'Exam',
             'location': 'block-v1:ORG2+1+1+type@homework+block@3',
         }
     }
 
     assert all(isinstance(value['location'], str) for _, value in serializer.grading_info.items())
     assert all(isinstance(key, str) for key in serializer.subsection_locations)
-
     assert not DeepDiff(serializer.grading_info, expected_grading_info, ignore_order=True)
     assert not DeepDiff(serializer.subsection_locations, {
-        'block-v1:ORG2+1+1+type@homework+block@1': '0',
-        'block-v1:ORG2+1+1+type@homework+block@2': '1',
-        'block-v1:ORG2+1+1+type@homework+block@3': '2',
+        'block-v1:ORG2+1+1+type@homework+block@1': 'course-v1:ORG2+1+1_0',
+        'block-v1:ORG2+1+1+type@homework+block@2': 'course-v1:ORG2+1+1_1',
+        'block-v1:ORG2+1+1+type@homework+block@3': 'course-v1:ORG2+1+1_2',
     }, ignore_order=True)
 
 
@@ -334,7 +361,7 @@ def test_learner_details_for_course_serializer_collect_grading_info_not_used(
 
     assert not serializer.grading_info
     assert not serializer.subsection_locations
-    serializer.collect_grading_info()
+    serializer.collect_grading_info(['course-v1:ORG2+1+1'])
     assert not serializer.grading_info
     assert not serializer.subsection_locations