feat: Add EnrollmentsService in XBlockRuntime and block renderer (#…

…34238)

* feat: Add EnrollmentsService in XBlockRuntime and block renderer

These changes give ability to use `EnrollmentsService` in XBlocks
Add `get_active_enrollments_by_course_and_user` method to `EnrollmentsService` which can be used to get active enrollment of user for a give course

lms/djangoapps/courseware/block_render.py

@@ -71,6 +71,7 @@
 from openedx.core.djangoapps.bookmarks.api import BookmarksService
 from openedx.core.djangoapps.crawlers.models import CrawlersConfig
 from openedx.core.djangoapps.credit.services import CreditService
+from openedx.core.djangoapps.enrollments.services import EnrollmentsService
 from openedx.core.djangoapps.util.user_utils import SystemUser
 from openedx.core.djangolib.markup import HTML
 from openedx.core.lib.api.authentication import BearerAuthenticationAllowInactiveUser
@@ -633,6 +634,7 @@ def inner_get_block(block: XBlock) -> XBlock | None:
         'teams_configuration': TeamsConfigurationService(),
         'call_to_action': CallToActionService(),
         'publish': EventPublishingService(user, course_id, track_function),
+        'enrollments': EnrollmentsService(),
     }
 
     runtime.get_block_for_descriptor = inner_get_block

openedx/core/djangoapps/enrollments/services.py

@@ -5,13 +5,16 @@
 from operator import or_
 
 from django.conf import settings
+from django.contrib.auth import get_user_model
 from django.db.models import Q
 
 from opaque_keys.edx.keys import CourseKey
 from common.djangoapps.course_modes.models import CourseMode
 from common.djangoapps.student.models import CourseEnrollment
 from openedx.core.djangoapps.content.course_overviews.api import get_course_overview_or_none
 
+USER_MODEL = get_user_model()
+
 
 class EnrollmentsService:
     """
@@ -25,6 +28,34 @@ def get_active_enrollments_by_course(self, course_id):
         """
         return CourseEnrollment.objects.filter(course_id=course_id, is_active=True)
 
+    def get_active_enrollments_by_course_and_user(self, course_id, user_id):
+        """
+        Returns a list of active enrollments for a course and user.
+
+        Parameters:
+        * course_id: course ID for the course
+        * user_id: ID of User object
+        """
+        user = self._get_user_by_id(user_id)
+        enrollment = CourseEnrollment.get_enrollment(user, course_id)
+        if not enrollment or not enrollment.is_active:
+            # not enrolled
+            return None
+
+        return enrollment
+
+    def _get_user_by_id(self, user_id):
+        """
+        Returns s User object for give ID
+
+        Parameters:
+        * user_id: ID of User object
+        """
+        try:
+            return USER_MODEL.objects.get(id=user_id)
+        except USER_MODEL.DoesNotExist:
+            return None
+
     def _get_enrollments_for_course_proctoring_eligible_modes(self, course_id, allow_honor_mode=False):
         """
         Return all enrollments for a course that are in a mode that makes the corresponding user

openedx/core/djangoapps/enrollments/tests/test_services.py

@@ -28,6 +28,7 @@ def setUp(self):
             CourseMode.PROFESSIONAL,
             CourseMode.VERIFIED
         ]
+        self.users = []
         self.course = CourseOverviewFactory.create(enable_proctored_exams=True)
 
         for index in range(len(self.course_modes)):
@@ -39,11 +40,51 @@ def setUp(self):
                 username=f'user{index}',
                 email=f'LEARNER{index}@example.com'
             )
+            self.users.append(user)
             CourseEnrollment.enroll(user, course_id, mode=course_mode)
 
     def enrollment_to_dict(self, enrollment):
         return {'username': enrollment.username, 'mode': enrollment.mode}
 
+    def test_get_active_enrollments_by_course(self):
+        """
+        Test that it returns a list of active enrollments for a course
+        """
+        enrollments = self.service.get_active_enrollments_by_course(course_id=str(self.course.id))
+        self.assertEqual(len(self.course_modes), len(enrollments))
+
+    def test_get_active_enrollments_by_course_and_user(self):
+        """
+        Test that it returns a list of active enrollments for a course and user
+        """
+        enrollment = self.service.get_active_enrollments_by_course_and_user(
+            str(self.course.id),
+            self.users[0].id,
+        )
+        expected_values = {'username': 'user0', 'mode': 'audit'}
+        self.assertDictEqual(self.enrollment_to_dict(enrollment), expected_values)
+
+    def test_get_active_enrollments_by_course_and_user_none_course(self):
+        """
+        Test that it returns a list of active enrollments for a course and user
+        for a non existing enrollment
+        """
+        enrollments = self.service.get_active_enrollments_by_course_and_user(
+            'course-v1:testx+none+123',
+            self.users[0].id,
+        )
+        self.assertIsNone(enrollments)
+
+    def test_get_active_enrollments_by_course_and_user_none_user(self):
+        """
+        Test getting enrollments for a course and non existing user raise error
+        """
+        with self.assertRaises(AssertionError):
+            self.service.get_active_enrollments_by_course_and_user(
+                str(self.course.id),
+                0000,
+            )
+
     def test_get_enrollments_can_take_proctored_exams_by_course(self):
         """
         Test that it returns a list of active enrollments

openedx/core/djangoapps/xblock/runtime/runtime.py

@@ -36,6 +36,7 @@
 from lms.djangoapps.courseware.model_data import DjangoKeyValueStore, FieldDataCache
 from lms.djangoapps.grades.api import signals as grades_signals
 from openedx.core.types import User as UserType
+from openedx.core.djangoapps.enrollments.services import EnrollmentsService
 from openedx.core.djangoapps.xblock.apps import get_xblock_app_config
 from openedx.core.djangoapps.xblock.data import StudentDataMode
 from openedx.core.djangoapps.xblock.runtime.ephemeral_field_data import EphemeralKeyValueStore
@@ -298,6 +299,8 @@ def service(self, block: XBlock, service_name: str):
             )
         elif service_name == 'publish':
             return EventPublishingService(self.user, context_key, make_track_function())
+        elif service_name == 'enrollments':
+            return EnrollmentsService()
 
         # Check if the XBlockRuntimeSystem wants to handle this:
         service = self.system.get_service(block, service_name)