feat: adds support for enrolling users in invite-only courses

enterprise/admin/__init__.py

@@ -204,7 +204,7 @@ class EnterpriseCustomerAdmin(DjangoObjectActions, SimpleHistoryAdmin):
         ('Integration and learning platform settings', {
             'fields': ('enable_portal_lms_configurations_screen', 'enable_portal_saml_configuration_screen',
                        'enable_slug_login', 'replace_sensitive_sso_username', 'hide_course_original_price',
-                       'enable_generation_of_api_credentials')
+                       'enable_generation_of_api_credentials', 'allow_enrollment_in_invite_only_courses')
         }),
         ('Recommended default settings for all enterprise customers', {
             'fields': ('site', 'customer_type', 'enable_learner_portal',

enterprise/admin/forms.py

@@ -395,6 +395,7 @@ class Meta:
             "enable_audit_data_reporting",
             "replace_sensitive_sso_username",
             "hide_course_original_price",
+            "allow_enrollment_in_invite_only_courses",
             "enable_portal_code_management_screen",
             "enable_portal_subscription_management_screen",
             "enable_learner_portal",

enterprise/api_client/lms.py

@@ -6,6 +6,7 @@
 import time
 from urllib.parse import urljoin
 
+import requests
 from opaque_keys.edx.keys import CourseKey
 from requests.exceptions import (  # pylint: disable=redefined-builtin
     ConnectionError,
@@ -274,6 +275,31 @@ def get_enrolled_courses(self, username):
         response.raise_for_status()
         return response.json()
 
+    def allow_enrollment(self, email: str, course_id: str, auto_enroll: bool = False):
+        """
+        Call the enrollment API to allow enrollment for the given email and course_id.
+        Args:
+            email (str): The email address of the user to be allowed to enroll in the course.
+            course_id (str): The string value of the course's unique identifier.
+            auto_enroll (bool): Whether to auto-enroll the user in the course upon registration / activation.
+        Returns:
+            dict: A dictionary containing details of the created CourseEnrollmentAllowed object.
+        """
+        api_url = self.get_api_url("enrollment_allowed")
+        response = self.client.post(
+            f"{api_url}/",
+            json={
+                'email': email,
+                'course_id': course_id,
+                'auto_enroll': auto_enroll,
+            }
+        )
+        if response.status_code == requests.codes.conflict:
+            LOGGER.info(response.json()["message"])
+        else:
+            response.raise_for_status()
+        return response.json()
+
 
 class CourseApiClient(NoAuthAPIClient):
     """

enterprise/migrations/0207_enterprisecustomer_allow_enrollment_in_invite_only_courses_and_more.py

@@ -0,0 +1,23 @@
+# Generated by Django 4.2.13 on 2024-12-06 03:48
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('enterprise', '0206_auto_20240408_1344'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='enterprisecustomer',
+            name='allow_enrollment_in_invite_only_courses',
+            field=models.BooleanField(default=False, help_text="Specifies if learners are allowed to enroll into courses marked as 'invitation-only', when they attempt to enroll from the landing page."),
+        ),
+        migrations.AddField(
+            model_name='historicalenterprisecustomer',
+            name='allow_enrollment_in_invite_only_courses',
+            field=models.BooleanField(default=False, help_text="Specifies if learners are allowed to enroll into courses marked as 'invitation-only', when they attempt to enroll from the landing page."),
+        ),
+    ]

enterprise/models.py

@@ -447,6 +447,14 @@ class Meta:
         help_text=_("Display Demo data from analyitcs and learner progress report for demo customer.")
     )
 
+    allow_enrollment_in_invite_only_courses = models.BooleanField(
+        default=False,
+        help_text=_(
+            "Specifies if learners are allowed to enroll into courses marked as 'invitation-only', "
+            "when they attempt to enroll from the landing page."
+        )
+    )
+
     contact_email = models.EmailField(
         verbose_name="Customer admin contact email:",
         null=True,

enterprise/utils.py

@@ -2028,6 +2028,8 @@ def enroll_subsidy_users_in_courses(enterprise_customer, subsidy_users_info, dis
                 [{ 'user_id': <lms_user_id>, 'email': <email>, 'course_run_key': <key> } ... ]
         }
     """
+    from enterprise.api_client.lms import EnrollmentApiClient  # pylint: disable=import-outside-toplevel
+    enrollment_api_client = EnrollmentApiClient()
     results = {
         'successes': [],
         'pending': [],
@@ -2042,6 +2044,7 @@ def enroll_subsidy_users_in_courses(enterprise_customer, subsidy_users_info, dis
         transaction_id = subsidy_user_info.get('transaction_id')
         activation_link = subsidy_user_info.get('activation_link')
         force_enrollment = subsidy_user_info.get('force_enrollment', False)
+        invitation_only = subsidy_user_info.get('invitation_only')
 
         if user_id and user_email:
             user = User.objects.filter(id=subsidy_user_info['user_id']).first()
@@ -2064,6 +2067,8 @@ def enroll_subsidy_users_in_courses(enterprise_customer, subsidy_users_info, dis
                 enrollment_source = enterprise_enrollment_source_model().get_source(
                     enterprise_enrollment_source_model().CUSTOMER_ADMIN
                 )
+                if invitation_only and enterprise_customer.allow_enrollment_in_invite_only_courses:
+                    ensure_course_enrollment_is_allowed(course_run_key, user.email, enrollment_api_client)
                 succeeded, created, source_uuid = customer_admin_enroll_user_with_status(
                     enterprise_customer,
                     user,
@@ -2102,6 +2107,8 @@ def enroll_subsidy_users_in_courses(enterprise_customer, subsidy_users_info, dis
                     discount=discount,
                     license_uuid=license_uuid
                 )
+                if invitation_only and enterprise_customer.allow_enrollment_in_invite_only_courses:
+                    ensure_course_enrollment_is_allowed(course_run_key, user_email, enrollment_api_client)
                 results['pending'].append({
                     'user': pending_user,
                     'email': user_email,
@@ -2455,3 +2462,17 @@ def get_integrations_for_customers(customer_uuid):
         if choice.objects.filter(enterprise_customer__uuid=customer_uuid, active=True):
             unique_integrations.append(code)
     return unique_integrations
+
+
+def ensure_course_enrollment_is_allowed(course_id: str, email: str, enrollment_api_client):
+    """
+    Calls the enrollment API to create a CourseEnrollmentAllowed object for
+    invitation-only courses.
+    Arguments:
+        course_id (str): ID of the course to allow enrollment
+        email (str): email of the user whose enrollment should be allowed
+        enrollment_api_client (:class:`enterprise.api_client.lms.EnrollmentApiClient`): Enrollment API Client
+    """
+    course_details = enrollment_api_client.get_course_details(course_id)
+    if course_details["invite_only"]:
+        enrollment_api_client.allow_enrollment(email, course_id)

enterprise/views.py

@@ -63,6 +63,7 @@
     CourseEnrollmentPermissionError,
     NotConnectedToOpenEdX,
     clean_html_for_template_rendering,
+    ensure_course_enrollment_is_allowed,
     filter_audit_course_modes,
     format_price,
     get_active_course_runs,
@@ -689,6 +690,14 @@ def _enroll_learner_in_course(
                             course_modes=course_mode
                         )
                     )
+                    if enterprise_customer.allow_enrollment_in_invite_only_courses:
+                        ensure_course_enrollment_is_allowed(course_id, request.user.email, enrollment_api_client)
+                        LOGGER.info(
+                            'User {user} is allowed to enroll in Course {course_id}.'.format(
+                                user=request.user.username,
+                                course_id=course_id
+                            )
+                        )
                     try:
                         enrollment_api_client.enroll_user_in_course(
                             request.user.username,

tests/test_enterprise/api_client/test_lms.py

@@ -449,6 +449,60 @@ def test_unenroll_already_unenrolled():
     assert not unenrolled
 
 
+@responses.activate
+@mock.patch('enterprise.api_client.client.JwtBuilder', mock.Mock())
+def test_allow_enrollment():
+    email = "[email protected]"
+    course_id = "course-v1:edX+DemoX+Demo_Course"
+    expected_response = {
+        "email": email,
+        "course_id": course_id,
+        "auto_enroll": False
+    }
+    responses.add(
+        responses.POST,
+        _url("enrollment", "enrollment_allowed/"),
+        json=expected_response,
+    )
+
+    client = lms_api.EnrollmentApiClient()
+    allowed = client.allow_enrollment(email, course_id)
+    assert allowed == expected_response
+
+
+@responses.activate
+@mock.patch('enterprise.api_client.client.JwtBuilder', mock.Mock())
+def test_allow_enrollment_raises_an_exception_on_error():
+    expected_response = {"message": "Bad Request"}
+    responses.add(
+        responses.POST,
+        _url("enrollment", "enrollment_allowed/"),
+        json=expected_response,
+        status=requests.codes.bad_request
+    )
+
+    client = lms_api.EnrollmentApiClient()
+    with raises(requests.HTTPError):
+        client.allow_enrollment("", "")
+
+
+@responses.activate
+@mock.patch('enterprise.api_client.client.JwtBuilder', mock.Mock())
+def test_allow_enrollment_does_not_raise_exception_on_conflict():
+    email = "[email protected]"
+    course_id = "course-v1:edX+DemoX+Demo_Course"
+    expected_response = {"message": f"An enrollment allowed with email {email} and course {course_id} already exists."}
+    responses.add(
+        responses.POST,
+        _url("enrollment", "enrollment_allowed/"),
+        json=expected_response,
+        status=requests.codes.conflict
+    )
+
+    client = lms_api.EnrollmentApiClient()
+    assert expected_response == client.allow_enrollment(email, course_id)
+
+
 @responses.activate
 def test_get_full_course_details():
     course_id = "course-v1:edX+DemoX+Demo_Course"

tests/test_enterprise/test_utils.py

@@ -17,6 +17,7 @@
 from enterprise.models import EnterpriseCourseEnrollment, LicensedEnterpriseCourseEnrollment
 from enterprise.utils import (
     enroll_subsidy_users_in_courses,
+    ensure_course_enrollment_is_allowed,
     get_default_invite_key_expiration_date,
     get_idiff_list,
     get_platform_logo_url,
@@ -525,6 +526,69 @@ def test_enroll_subsidy_users_in_courses_user_identifier_failures(
         )
         self.assertEqual(len(EnterpriseCourseEnrollment.objects.all()), 0)
 
+    @ddt.unpack
+    @ddt.data(
+        (True, True),
+        (True, False),
+        (False, True),
+        (False, False),
+    )
+    @mock.patch('enterprise.utils.lms_update_or_create_enrollment')
+    @mock.patch('enterprise.utils.ensure_course_enrollment_is_allowed')
+    def test_enroll_subsidy_users_in_courses_for_invite_only_courses(
+        self,
+        invite_only,
+        enrollment_allowed,
+        mock_ensure_course_enrollment_is_allowed,
+        mock_update_or_create_enrollment,
+    ):
+        """
+        Test that the users ensure_course_enrollemnt_is_allowed is called for
+        invitiation-only courses when the enterprise_customer has the flag enabled.
+        """
+        self.create_user()
+
+        ent_customer = factories.EnterpriseCustomerFactory(
+            uuid=FAKE_UUIDS[0],
+            name="test_enterprise",
+            allow_enrollment_in_invite_only_courses=enrollment_allowed,
+        )
+        factories.EnterpriseCustomerUserFactory(
+            user_id=self.user.id,
+            enterprise_customer=ent_customer,
+        )
+        licensed_users_info = [{
+            'email': self.user.email,
+            'course_run_key': 'course-key-v1',
+            'course_mode': 'verified',
+            'license_uuid': '5b77bdbade7b4fcb838f8111b68e18ae',
+            'invitation_only': invite_only,
+        }]
+
+        mock_update_or_create_enrollment.return_value = True
+        result = enroll_subsidy_users_in_courses(ent_customer, licensed_users_info)
+        self.assertEqual(
+            {
+                'pending': [],
+                'successes': [{
+                    'user_id': self.user.id,
+                    'email': self.user.email,
+                    'course_run_key': 'course-key-v1',
+                    'user': self.user,
+                    'created': True,
+                    'activation_link': None,
+                    'enterprise_fulfillment_source_uuid': LicensedEnterpriseCourseEnrollment.objects.first().uuid,
+                }],
+                'failures': []
+            },
+            result
+        )
+        self.assertEqual(len(EnterpriseCourseEnrollment.objects.all()), 1)
+        if invite_only and enrollment_allowed:
+            mock_ensure_course_enrollment_is_allowed.assert_called()
+        else:
+            mock_ensure_course_enrollment_is_allowed.assert_not_called()
+
     def test_enroll_pending_licensed_users_in_courses_succeeds(self):
         """
         Test that users that do not exist are pre-enrolled by enroll_subsidy_users_in_courses and returned under the
@@ -650,3 +714,22 @@ def test_truncate_string(self):
         (truncated_string, was_truncated) = truncate_string(test_string_2)
         self.assertTrue(was_truncated)
         self.assertEqual(len(truncated_string), MAX_ALLOWED_TEXT_LENGTH)
+
+    @ddt.data(True, False)
+    def test_ensure_course_enrollment_is_allowed(self, invite_only):
+        """
+        Test that the enrollment allow endpoint is called for the "invite_only" courses.
+        """
+        self.create_user()
+        mock_enrollment_api = mock.Mock()
+        mock_enrollment_api.get_course_details.return_value = {"invite_only": invite_only}
+
+        ensure_course_enrollment_is_allowed("test-course-id", self.user.email, mock_enrollment_api)
+
+        if invite_only:
+            mock_enrollment_api.allow_enrollment.assert_called_with(
+                self.user.email,
+                "test-course-id",
+            )
+        else:
+            mock_enrollment_api.allow_enrollment.assert_not_called()