diff --git a/lms/djangoapps/courseware/block_render.py b/lms/djangoapps/courseware/block_render.py index 5afcaac044af..1bae90322487 100644 --- a/lms/djangoapps/courseware/block_render.py +++ b/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 diff --git a/openedx/core/djangoapps/enrollments/services.py b/openedx/core/djangoapps/enrollments/services.py index e948ebac699e..34bdad9d2107 100644 --- a/openedx/core/djangoapps/enrollments/services.py +++ b/openedx/core/djangoapps/enrollments/services.py @@ -5,6 +5,7 @@ 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 @@ -12,6 +13,8 @@ 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 diff --git a/openedx/core/djangoapps/enrollments/tests/test_services.py b/openedx/core/djangoapps/enrollments/tests/test_services.py index 1821b6a66461..46cdf057dc25 100644 --- a/openedx/core/djangoapps/enrollments/tests/test_services.py +++ b/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 diff --git a/openedx/core/djangoapps/xblock/runtime/runtime.py b/openedx/core/djangoapps/xblock/runtime/runtime.py index 0b6e7a9ce2f8..09f5fa93a0e2 100644 --- a/openedx/core/djangoapps/xblock/runtime/runtime.py +++ b/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)