From 2be191dcb96d269f8e043353e8a1abb54d8de328 Mon Sep 17 00:00:00 2001
From: Arunmozhi <arunmozhi@opencraft.com>
Date: Wed, 24 May 2023 17:30:23 +0530
Subject: [PATCH 1/9] feat: enable enrolling in invite_only courses via manage
 learners UI

---
 enterprise/admin/forms.py                     | 261 ++++++++++++------
 enterprise/admin/views.py                     |  12 +-
 enterprise/api_client/lms.py                  |  13 +-
 enterprise/models.py                          |  16 +-
 .../static/enterprise/js/manage_learners.js   |  36 ++-
 enterprise/utils.py                           |  12 +-
 test_utils/fake_enrollment_api.py             |   2 +-
 tests/test_admin/test_view.py                 | 132 +++++++--
 tests/test_enterprise/api/test_views.py       |  62 +++--
 tests/test_enterprise/api_client/test_lms.py  |   3 +-
 tests/test_enterprise/test_utils.py           |   6 +-
 11 files changed, 408 insertions(+), 147 deletions(-)

diff --git a/enterprise/admin/forms.py b/enterprise/admin/forms.py
index fddb2408ee..b1a4c9a2a1 100644
--- a/enterprise/admin/forms.py
+++ b/enterprise/admin/forms.py
@@ -19,7 +19,10 @@
 from django.utils.translation import gettext as _
 
 from enterprise import utils
-from enterprise.admin.utils import email_or_username__to__email, split_usernames_and_emails
+from enterprise.admin.utils import (
+    email_or_username__to__email,
+    split_usernames_and_emails,
+)
 from enterprise.admin.widgets import SubmitInput
 from enterprise.models import (
     AdminNotification,
@@ -34,7 +37,9 @@
 from enterprise.utils import ValidationMessages, validate_email_to_link
 
 try:
-    from common.djangoapps.third_party_auth.models import SAMLProviderConfig as saml_provider_configuration
+    from common.djangoapps.third_party_auth.models import (
+        SAMLProviderConfig as saml_provider_configuration,
+    )
 except ImportError:
     saml_provider_configuration = None
 
@@ -46,10 +51,11 @@ class ManageLearnersForm(forms.Form):
     """
     Form to manage learner additions.
     """
+
     email_or_username = forms.CharField(
-        label=_(
-            "To add a single learner, enter an email address or username."),
-        required=False)
+        label=_("To add a single learner, enter an email address or username."),
+        required=False,
+    )
     bulk_upload_csv = forms.FileField(
         label=_(
             "To add multiple learners, upload a .csv file that contains a "
@@ -61,15 +67,25 @@ class ManageLearnersForm(forms.Form):
             "by the heading 'email' in the first row. Optionally, the .csv file may contain "
             "a column of course run keys, indicated by the heading 'course_id' in the first row, to "
             "enroll learners in multiple courses."
-        )
+        ),
     )
     course = forms.CharField(
-        label=_("Enroll these learners in this course"), required=False,
+        label=_("Enroll these learners in this course"),
+        required=False,
         help_text=_("To enroll learners in a course, enter a course ID."),
     )
+    force_enrollment = forms.BooleanField(
+        label=_("Force Enrollment"),
+        help_text=_(
+            "The selected course is 'Invite Only'. Only staff can enroll learners to this course."
+        ),
+        required=False,
+    )
     course_mode = forms.ChoiceField(
-        label=_("Course enrollment track"), required=False,
-        choices=BLANK_CHOICE_DASH + [
+        label=_("Course enrollment track"),
+        required=False,
+        choices=BLANK_CHOICE_DASH
+        + [
             ("audit", _("Audit")),
             ("verified", _("Verified")),
             ("professional", _("Professional Education")),
@@ -79,22 +95,27 @@ class ManageLearnersForm(forms.Form):
         ],
     )
     reason = forms.CharField(label=_("Reason for manual enrollment"), required=False)
-    sales_force_id = forms.CharField(label=_("Salesforce Opportunity ID"), required=False)
+    sales_force_id = forms.CharField(
+        label=_("Salesforce Opportunity ID"), required=False
+    )
     discount = forms.DecimalField(
         label=_("Discount percentage for manual enrollment"),
         help_text=_("Discount percentage should be from 0 to 100"),
         required=True,
         decimal_places=5,
-        initial=0.0
+        initial=0.0,
     )
 
     class NotificationTypes:
         """
         Namespace class for notification types
         """
-        BY_EMAIL = 'by_email'
-        NO_NOTIFICATION = 'do_not_notify'
-        DEFAULT = getattr(settings, 'DEFAULT_ENTERPRISE_NOTIFICATION_MECHANISM', BY_EMAIL)
+
+        BY_EMAIL = "by_email"
+        NO_NOTIFICATION = "do_not_notify"
+        DEFAULT = getattr(
+            settings, "DEFAULT_ENTERPRISE_NOTIFICATION_MECHANISM", BY_EMAIL
+        )
 
     notify_on_enrollment = forms.ChoiceField(
         label=_("Notify learners of enrollment"),
@@ -110,6 +131,7 @@ class Modes:
         """
         Namespace class for form modes.
         """
+
         MODE_SINGULAR = "singular"
         MODE_BULK = "bulk"
 
@@ -117,6 +139,7 @@ class Fields:
         """
         Namespace class for field names.
         """
+
         GENERAL_ERRORS = forms.forms.NON_FIELD_ERRORS
 
         EMAIL_OR_USERNAME = "email_or_username"
@@ -128,11 +151,13 @@ class Fields:
         REASON = "reason"
         SALES_FORCE_ID = "sales_force_id"
         DISCOUNT = "discount"
+        FORCE_ENROLLMENT = "force_enrollment"
 
     class CsvColumns:
         """
         Namespace class for CSV column names.
         """
+
         EMAIL = "email"
         COURSE_ID = "course_id"
 
@@ -145,9 +170,9 @@ def __init__(self, *args, **kwargs):
             user (django.contrib.auth.models.User): current user
             enterprise_customer (enterprise.models.EnterpriseCustomer): current customer
         """
-        user = kwargs.pop('user', None)
+        user = kwargs.pop("user", None)
         self._user = user
-        self._enterprise_customer = kwargs.pop('enterprise_customer', None)
+        self._enterprise_customer = kwargs.pop("enterprise_customer", None)
         super().__init__(*args, **kwargs)
 
     def clean_email_or_username(self):
@@ -202,8 +227,12 @@ def clean_course(self):
         course_id = self.cleaned_data[self.Fields.COURSE].strip()
         if not course_id:
             return None
-        enterprise_customer = utils.get_enterprise_customer(self._enterprise_customer.uuid)
-        course_details = utils.validate_course_exists_for_enterprise(enterprise_customer, course_id)
+        enterprise_customer = utils.get_enterprise_customer(
+            self._enterprise_customer.uuid
+        )
+        course_details = utils.validate_course_exists_for_enterprise(
+            enterprise_customer, course_id
+        )
         return course_details
 
     def clean_reason(self):
@@ -260,11 +289,15 @@ def _validate_bulk_upload(self):
         bulk_upload_csv = self.cleaned_data.get(self.Fields.BULK_UPLOAD)
         if bulk_upload_csv:
             bulk_csv_contents = bulk_upload_csv.read()
-            csv_column_names = bulk_csv_contents.decode('utf-8').split('\n', 1)[0].split(',')
+            csv_column_names = (
+                bulk_csv_contents.decode("utf-8").split("\n", 1)[0].split(",")
+            )
             if self.CsvColumns.COURSE_ID in csv_column_names:
                 # course id column exists in csv, so validate conditionally required fields
                 if self.cleaned_data.get(self.Fields.COURSE):
-                    raise ValidationError(ValidationMessages.BOTH_COURSE_FIELDS_SPECIFIED)
+                    raise ValidationError(
+                        ValidationMessages.BOTH_COURSE_FIELDS_SPECIFIED
+                    )
                 if not self.cleaned_data.get(self.Fields.COURSE_MODE):
                     raise ValidationError(ValidationMessages.COURSE_WITHOUT_COURSE_MODE)
                 if not self.cleaned_data.get(self.Fields.REASON):
@@ -282,10 +315,12 @@ def _validate_course(self):
                 raise ValidationError(ValidationMessages.COURSE_WITHOUT_COURSE_MODE)
             valid_course_modes = course_details["course_modes"]
             if all(course_mode != mode["slug"] for mode in valid_course_modes):
-                error = ValidationError(ValidationMessages.COURSE_MODE_INVALID_FOR_COURSE.format(
-                    course_mode=course_mode,
-                    course_id=course_details["course_id"],
-                ))
+                error = ValidationError(
+                    ValidationMessages.COURSE_MODE_INVALID_FOR_COURSE.format(
+                        course_mode=course_mode,
+                        course_id=course_details["course_id"],
+                    )
+                )
                 raise ValidationError({self.Fields.COURSE_MODE: error})
 
     def _validate_reason(self):
@@ -304,10 +339,11 @@ class ManageLearnersDataSharingConsentForm(forms.Form):
     """
     Form to request DSC from a learner.
     """
+
     email_or_username = forms.CharField(
         label=_("Email/Username"),
         help_text=_("Enter an email address or username."),
-        required=True
+        required=True,
     )
     course = forms.CharField(
         label=_("Course"),
@@ -319,6 +355,7 @@ class Fields:
         """
         Namespace class for field names.
         """
+
         EMAIL_OR_USERNAME = "email_or_username"
         COURSE = "course"
 
@@ -329,7 +366,7 @@ def __init__(self, *args, **kwargs):
         Arguments:
             enterprise_customer (enterprise.models.EnterpriseCustomer): current customer
         """
-        self._enterprise_customer = kwargs.pop('enterprise_customer', None)
+        self._enterprise_customer = kwargs.pop("enterprise_customer", None)
         super().__init__(*args, **kwargs)
 
     def clean_course(self):
@@ -339,7 +376,9 @@ def clean_course(self):
         course_id = self.cleaned_data[self.Fields.COURSE].strip()
         if not course_id:
             return None
-        enterprise_customer = utils.get_enterprise_customer(self._enterprise_customer.uuid)
+        enterprise_customer = utils.get_enterprise_customer(
+            self._enterprise_customer.uuid
+        )
         utils.validate_course_exists_for_enterprise(enterprise_customer, course_id)
         return course_id
 
@@ -362,7 +401,9 @@ def is_course_in_catalog(self, course_id):
         """
         Check whether course exists in enterprise customer catalog.
         """
-        enterprise_customer = utils.get_enterprise_customer(self._enterprise_customer.uuid)
+        enterprise_customer = utils.get_enterprise_customer(
+            self._enterprise_customer.uuid
+        )
         return enterprise_customer.catalog_contains_course(course_id)
 
     def is_user_linked(self, email):
@@ -370,13 +411,16 @@ def is_user_linked(self, email):
         Check whether user is linked to the enterprise customer or not.
         """
         user = User.objects.get(email=email)
-        return utils.get_enterprise_customer_user(user.id, self._enterprise_customer.uuid)
+        return utils.get_enterprise_customer_user(
+            user.id, self._enterprise_customer.uuid
+        )
 
 
 class EnterpriseCustomerAdminForm(forms.ModelForm):
     """
     Alternate form for the EnterpriseCustomer admin page.
     """
+
     class Meta:
         model = EnterpriseCustomer
         fields = (
@@ -414,14 +458,19 @@ class Meta:
 
 class EnterpriseCustomerCatalogAdminForm(forms.ModelForm):
     """
-        form for EnterpriseCustomerCatalogAdmin class.
+    form for EnterpriseCustomerCatalogAdmin class.
     """
+
     class Meta:
         model = EnterpriseCustomerCatalog
         fields = "__all__"
 
-    preview_button = forms.Field(required=False, label='Actions', widget=SubmitInput(attrs={'value': _('Preview')}),
-                                 help_text=_("Hold Ctrl when clicking on button to open Preview in new tab"))
+    preview_button = forms.Field(
+        required=False,
+        label="Actions",
+        widget=SubmitInput(attrs={"value": _("Preview")}),
+        help_text=_("Hold Ctrl when clicking on button to open Preview in new tab"),
+    )
 
     @staticmethod
     def get_catalog_preview_uuid(post_data):
@@ -431,16 +480,22 @@ def get_catalog_preview_uuid(post_data):
 
         e.g: 'enterprise_customer_catalogs-0-preview_button'
         """
-        preview_button_expression = re.compile(r'enterprise_customer_catalogs-\d+-preview_button')
-        clicked_button_index_expression = re.compile(r'-(.+?)-')
+        preview_button_expression = re.compile(
+            r"enterprise_customer_catalogs-\d+-preview_button"
+        )
+        clicked_button_index_expression = re.compile(r"-(.+?)-")
         count = 0
         preview_button_index = None
         for key, _ in post_data.items():
             if preview_button_expression.match(key):
                 count += 1
-                preview_button_index = clicked_button_index_expression.search(key).group(1)
+                preview_button_index = clicked_button_index_expression.search(
+                    key
+                ).group(1)
         if count == 1:
-            return post_data.get('enterprise_customer_catalogs-' + preview_button_index + '-uuid')
+            return post_data.get(
+                "enterprise_customer_catalogs-" + preview_button_index + "-uuid"
+            )
         return None
 
 
@@ -451,6 +506,7 @@ class EnterpriseCustomerIdentityProviderAdminForm(forms.ModelForm):
     This form fetches identity providers from lms third_party_auth app.
     If third_party_auth app is not avilable it displays provider_id as a CharField.
     """
+
     class Meta:
         model = EnterpriseCustomerIdentityProvider
         fields = "__all__"
@@ -463,28 +519,34 @@ def __init__(self, *args, **kwargs):
         """
         super().__init__(*args, **kwargs)
         idp_choices = utils.get_idp_choices()
-        help_text = ''
+        help_text = ""
         if saml_provider_configuration:
             provider_id = self.instance.provider_id
-            url = reverse('admin:{}_{}_add'.format(
-                saml_provider_configuration._meta.app_label,
-                saml_provider_configuration._meta.model_name))
+            url = reverse(
+                "admin:{}_{}_add".format(
+                    saml_provider_configuration._meta.app_label,
+                    saml_provider_configuration._meta.model_name,
+                )
+            )
             if provider_id:
                 identity_provider = utils.get_identity_provider(provider_id)
                 if identity_provider:
-                    update_url = url + '?source={}'.format(identity_provider.pk)
-                    help_text = '<p><a href="{update_url}" target="_blank">View "{identity_provider}" details</a><p>'.\
-                        format(update_url=update_url, identity_provider=identity_provider.name)
+                    update_url = url + "?source={}".format(identity_provider.pk)
+                    help_text = '<p><a href="{update_url}" target="_blank">View "{identity_provider}" details</a><p>'.format(
+                        update_url=update_url, identity_provider=identity_provider.name
+                    )
                 else:
                     help_text += '<p style="margin-top:-5px;"> Make sure you have added a valid provider_id.</p>'
             else:
-                help_text += '<p style="margin-top:-5px;"><a target="_blank" href={add_url}>' \
-                             'Create a new identity provider</a></p>'.format(add_url=url)
+                help_text += (
+                    '<p style="margin-top:-5px;"><a target="_blank" href={add_url}>'
+                    "Create a new identity provider</a></p>".format(add_url=url)
+                )
 
         if idp_choices is not None:
-            self.fields['provider_id'] = forms.TypedChoiceField(
+            self.fields["provider_id"] = forms.TypedChoiceField(
                 choices=idp_choices,
-                label=_('Identity Provider'),
+                label=_("Identity Provider"),
                 help_text=mark_safe(help_text),
             )
 
@@ -496,8 +558,8 @@ def clean(self):
         """
         super().clean()
 
-        provider_id = self.cleaned_data.get('provider_id', None)
-        enterprise_customer = self.cleaned_data.get('enterprise_customer', None)
+        provider_id = self.cleaned_data.get("provider_id", None)
+        enterprise_customer = self.cleaned_data.get("enterprise_customer", None)
 
         if provider_id is None or enterprise_customer is None:
             # field validation for either provider_id or enterprise_customer has already raised
@@ -532,17 +594,23 @@ def clean(self):
             )
 
         # There should be always one default provider in case an enterprise has multiple providers.
-        identity_provider_total_forms = int(self.data.get('enterprise_customer_identity_providers-TOTAL_FORMS'))
+        identity_provider_total_forms = int(
+            self.data.get("enterprise_customer_identity_providers-TOTAL_FORMS")
+        )
         default_provider_values = []
         for form_num in range(identity_provider_total_forms):
             default_provider_values.append(
-                self.data.get('enterprise_customer_identity_providers-{}-default_provider'.format(form_num))
+                self.data.get(
+                    "enterprise_customer_identity_providers-{}-default_provider".format(
+                        form_num
+                    )
+                )
             )
-        providers_marked_default = default_provider_values.count('on')
+        providers_marked_default = default_provider_values.count("on")
         if providers_marked_default > 1:
-            raise ValidationError('Please select only one default provider.')
+            raise ValidationError("Please select only one default provider.")
         if identity_provider_total_forms > 1 and providers_marked_default == 0:
-            raise ValidationError('Please select one default provider.')
+            raise ValidationError("Please select one default provider.")
 
 
 class EnterpriseCustomerReportingConfigAdminForm(forms.ModelForm):
@@ -552,6 +620,7 @@ class EnterpriseCustomerReportingConfigAdminForm(forms.ModelForm):
     This form uses the PasswordInput widget to obscure passwords as they are
     being entered by the user.
     """
+
     enterprise_customer_catalogs = forms.ModelMultipleChoiceField(
         EnterpriseCustomerCatalog.objects.all(),
         required=False,
@@ -582,8 +651,8 @@ class Meta:
             "enterprise_customer_catalogs",
         )
         widgets = {
-            'decrypted_password': forms.widgets.PasswordInput(),
-            'decrypted_sftp_password': forms.widgets.PasswordInput(),
+            "decrypted_password": forms.widgets.PasswordInput(),
+            "decrypted_sftp_password": forms.widgets.PasswordInput(),
         }
 
     def clean(self):
@@ -591,45 +660,47 @@ def clean(self):
         Override of clean method to perform additional validation
         """
         cleaned_data = super().clean()
-        report_customer = cleaned_data.get('enterprise_customer')
-        data_type = cleaned_data.get('data_type')
+        report_customer = cleaned_data.get("enterprise_customer")
+        data_type = cleaned_data.get("data_type")
 
-        if data_type in EnterpriseCustomerReportingConfiguration.PEARSON_ONLY_REPORTS \
-                and report_customer.name != 'Pearson':
+        if (
+            data_type in EnterpriseCustomerReportingConfiguration.PEARSON_ONLY_REPORTS
+            and report_customer.name != "Pearson"
+        ):
             message = _(
                 'This data_type "{data_type}" is not supported for enterprise'
-                'customer {enterprise_customer}. Please select a different data_type.',
+                "customer {enterprise_customer}. Please select a different data_type.",
             ).format(
                 enterprise_customer=report_customer,
                 data_type=data_type,
             )
-            self.add_error('data_type', message)
+            self.add_error("data_type", message)
 
         # Check that any selected catalogs are tied to the selected enterprise.
         invalid_catalogs = [
-            '{} ({})'.format(catalog.title, catalog.uuid)
-            for catalog in cleaned_data.get('enterprise_customer_catalogs')
+            "{} ({})".format(catalog.title, catalog.uuid)
+            for catalog in cleaned_data.get("enterprise_customer_catalogs")
             if catalog.enterprise_customer != report_customer
         ]
 
         if invalid_catalogs:
             message = _(
-                'These catalogs for reporting do not match enterprise'
-                'customer {enterprise_customer}: {invalid_catalogs}',
+                "These catalogs for reporting do not match enterprise"
+                "customer {enterprise_customer}: {invalid_catalogs}",
             ).format(
                 enterprise_customer=report_customer,
                 invalid_catalogs=invalid_catalogs,
             )
-            self.add_error('enterprise_customer_catalogs', message)
+            self.add_error("enterprise_customer_catalogs", message)
 
 
 class TransmitEnterpriseCoursesForm(forms.Form):
     """
     Form to transmit courses metadata for enterprise customers.
     """
+
     channel_worker_username = forms.CharField(
-        label=_('Enter enterprise channel worker username.'),
-        required=True
+        label=_("Enter enterprise channel worker username."), required=True
     )
 
     def clean_channel_worker_username(self):
@@ -639,7 +710,7 @@ def clean_channel_worker_username(self):
         Returns:
             str: the cleaned value of channel user username for transmitting courses metadata.
         """
-        channel_worker_username = self.cleaned_data['channel_worker_username'].strip()
+        channel_worker_username = self.cleaned_data["channel_worker_username"].strip()
 
         try:
             User.objects.get(username=channel_worker_username)
@@ -661,10 +732,10 @@ class SystemWideEnterpriseUserRoleAssignmentForm(UserRoleAssignmentAdminForm):
     class Meta:
         model = SystemWideEnterpriseUserRoleAssignment
         fields = [
-            'user',
-            'role',
-            'enterprise_customer',
-            'applies_to_all_contexts',
+            "user",
+            "role",
+            "enterprise_customer",
+            "applies_to_all_contexts",
         ]
 
     def __init__(self, *args, **kwargs):
@@ -675,11 +746,15 @@ def __init__(self, *args, **kwargs):
         """
         super().__init__(*args, **kwargs)
         try:
-            self.fields['enterprise_customer'].queryset = EnterpriseCustomer.objects.filter(
+            self.fields[
+                "enterprise_customer"
+            ].queryset = EnterpriseCustomer.objects.filter(
                 enterprise_customer_users__user_id=self.instance.user.id,
             )
         except SystemWideEnterpriseUserRoleAssignment.user.RelatedObjectDoesNotExist:
-            self.fields['enterprise_customer'].queryset = EnterpriseCustomer.objects.none()
+            self.fields[
+                "enterprise_customer"
+            ].queryset = EnterpriseCustomer.objects.none()
 
 
 class EnterpriseFeatureUserRoleAssignmentForm(UserRoleAssignmentAdminForm):
@@ -689,7 +764,7 @@ class EnterpriseFeatureUserRoleAssignmentForm(UserRoleAssignmentAdminForm):
 
     class Meta:
         model = EnterpriseFeatureUserRoleAssignment
-        fields = ['user', 'role']
+        fields = ["user", "role"]
 
 
 class AdminNotificationForm(forms.ModelForm):
@@ -699,9 +774,9 @@ class AdminNotificationForm(forms.ModelForm):
 
     class Meta:
         model = AdminNotification
-        fields = '__all__'
+        fields = "__all__"
         widgets = {
-            'text': forms.Textarea(attrs={'cols': 80, 'rows': 5}),
+            "text": forms.Textarea(attrs={"cols": 80, "rows": 5}),
         }
 
     def clean(self):
@@ -712,28 +787,32 @@ def clean(self):
         """
         cleaned_data = super().clean()
 
-        start_date = cleaned_data.get('start_date', None)
-        expiration_date = cleaned_data.get('expiration_date', None)
+        start_date = cleaned_data.get("start_date", None)
+        expiration_date = cleaned_data.get("expiration_date", None)
 
         if start_date is None or expiration_date is None:
             # Start and expiration dates are mandatory.
-            raise ValidationError('Start and expiration dates are mandatory.')
+            raise ValidationError("Start and expiration dates are mandatory.")
 
         if expiration_date < start_date:
             # `start_date` must always come before the `expiration_date`
-            raise ValidationError({'expiration_date': ['Expiration date should be after start date.']})
-        admin_notification = AdminNotification.objects.filter(
-            Q(start_date__range=(start_date, expiration_date)) |
-            Q(expiration_date__range=(start_date, expiration_date)) |
-            Q(start_date__lt=start_date, expiration_date__gt=expiration_date)
-        ).exclude(
-            pk=self.instance.id
-        ).exists()
+            raise ValidationError(
+                {"expiration_date": ["Expiration date should be after start date."]}
+            )
+        admin_notification = (
+            AdminNotification.objects.filter(
+                Q(start_date__range=(start_date, expiration_date))
+                | Q(expiration_date__range=(start_date, expiration_date))
+                | Q(start_date__lt=start_date, expiration_date__gt=expiration_date)
+            )
+            .exclude(pk=self.instance.id)
+            .exists()
+        )
 
         if admin_notification:
             # This should not happen, as there can be only one admin notification instance in a particular date range.
             message = _(
-                'Please select different date range. There is another notification scheduled in this date range.',
+                "Please select different date range. There is another notification scheduled in this date range.",
             )
             logger.exception(message)
 
diff --git a/enterprise/admin/views.py b/enterprise/admin/views.py
index 80cf8716ec..ee2a45fdbb 100644
--- a/enterprise/admin/views.py
+++ b/enterprise/admin/views.py
@@ -641,7 +641,8 @@ def _enroll_users(
             notify=True,
             enrollment_reason=None,
             sales_force_id=None,
-            discount=0.0
+            discount=0.0,
+            force_enrollment=False,
     ):
         """
         Enroll the users with the given email addresses to the course.
@@ -654,6 +655,7 @@ def _enroll_users(
             mode: The enrollment mode the users will be enrolled in the course with
             course_id: The ID of the course in which we want to enroll
             notify: Whether to notify (by email) the users that have been enrolled
+            force_enrollment: Force enrollment into "Invite Only" courses
         """
         pending_messages = []
         paid_modes = ['verified', 'professional']
@@ -667,6 +669,7 @@ def _enroll_users(
             enrollment_reason=enrollment_reason,
             discount=discount,
             sales_force_id=sales_force_id,
+            force_enrollment=force_enrollment,
         )
         all_successes = succeeded + pending
         if notify:
@@ -783,6 +786,7 @@ def post(self, request, customer_uuid):
             sales_force_id = manage_learners_form.cleaned_data.get(ManageLearnersForm.Fields.SALES_FORCE_ID)
             course_mode = manage_learners_form.cleaned_data.get(ManageLearnersForm.Fields.COURSE_MODE)
             course_id = None
+            force_enrollment = manage_learners_form.cleaned_data.get(ManageLearnersForm.Fields.FORCE_ENROLLMENT)
 
             if not course_id_with_emails:
                 course_details = manage_learners_form.cleaned_data.get(ManageLearnersForm.Fields.COURSE) or {}
@@ -797,7 +801,8 @@ def post(self, request, customer_uuid):
                         notify=notify,
                         enrollment_reason=manual_enrollment_reason,
                         sales_force_id=sales_force_id,
-                        discount=discount
+                        discount=discount,
+                        force_enrollment=force_enrollment,
                     )
             else:
                 for course_id, emails in course_id_with_emails.items():
@@ -812,7 +817,8 @@ def post(self, request, customer_uuid):
                             notify=notify,
                             enrollment_reason=manual_enrollment_reason,
                             sales_force_id=sales_force_id,
-                            discount=discount
+                            discount=discount,
+                            force_enrollment=force_enrollment,
                         )
 
             # Redirect to GET if everything went smooth.
diff --git a/enterprise/api_client/lms.py b/enterprise/api_client/lms.py
index c0114c3ae4..3aa681d65a 100644
--- a/enterprise/api_client/lms.py
+++ b/enterprise/api_client/lms.py
@@ -230,7 +230,15 @@ def has_course_mode(self, course_run_id, mode):
         return any(course_mode for course_mode in course_modes if course_mode['slug'] == mode)
 
     @JwtLmsApiClient.refresh_token
-    def enroll_user_in_course(self, username, course_id, mode, cohort=None, enterprise_uuid=None):
+    def enroll_user_in_course(
+        self,
+        username,
+        course_id,
+        mode,
+        cohort=None,
+        enterprise_uuid=None,
+        force_enrollment=False,
+    ):
         """
         Call the enrollment API to enroll the user in the course specified by course_id.
 
@@ -252,7 +260,8 @@ def enroll_user_in_course(self, username, course_id, mode, cohort=None, enterpri
                 'is_active': True,
                 'mode': mode,
                 'cohort': cohort,
-                'enterprise_uuid': str(enterprise_uuid)
+                'enterprise_uuid': str(enterprise_uuid),
+                'force_enrollment': force_enrollment,
             }
         )
 
diff --git a/enterprise/models.py b/enterprise/models.py
index 672d16a588..f2e0615f36 100644
--- a/enterprise/models.py
+++ b/enterprise/models.py
@@ -75,7 +75,7 @@
 )
 
 try:
-    from common.djangoapps.student.models import CourseEnrollment
+    from common.djangoapps.student.models import CourseEnrollment, CourseEnrollmentAllowed
 except ImportError:
     CourseEnrollment = None
 
@@ -647,7 +647,21 @@ def enroll_user_pending_registration_with_status(self, email, course_mode, *cour
             license_uuid = None
 
         new_enrollments = {}
+        enrollment_api_client = EnrollmentApiClient()
+
         for course_id in course_ids:
+            # Check if the course is "Invite Only" and add CEA if it is.
+            course_details = enrollment_api_client.get_course_details(course_id)
+
+            if course_details["invite_only"]:
+                if not CourseEnrollmentAllowed:
+                    raise NotConnectedToOpenEdX()
+
+                CourseEnrollmentAllowed.objects.update_or_create(
+                    email=email,
+                    course_id=course_id
+                )
+
             __, created = PendingEnrollment.objects.update_or_create(
                 user=pending_ecu,
                 course_id=course_id,
diff --git a/enterprise/static/enterprise/js/manage_learners.js b/enterprise/static/enterprise/js/manage_learners.js
index 5b12d4ad0b..940092467b 100644
--- a/enterprise/static/enterprise/js/manage_learners.js
+++ b/enterprise/static/enterprise/js/manage_learners.js
@@ -9,7 +9,7 @@ function makeOption(name, value) {
     return $("<option></option>").text(name).val(value);
 }
 
-function fillModeDropdown(data) {
+function updateCourseData(data) {
     /*
      Given a set of data fetched from the enrollment API, populate the Course Mode
      dropdown with those options that are valid for the course entered in the
@@ -19,6 +19,11 @@ function fillModeDropdown(data) {
     var previous_value = $course_mode.val();
     applyModes(data.course_modes);
     $course_mode.val(previous_value);
+
+    // If the course is invite-only, show the force enrollment box.
+    if (data.invite_only) {
+        $("#id_force_enrollment").parent().show();
+    }
 }
 
 function applyModes(modes) {
@@ -43,7 +48,7 @@ function loadCourseModes(success, failure) {
             return;
         }
         $.ajax({method: 'get', url: enrollmentApiRoot + "course/" + courseId})
-            .done(success || fillModeDropdown)
+            .done(success || updateCourseData)
             .fail(failure || function (err, jxHR, errstat) { disableMode(disableReason); });
     });
 }
@@ -134,11 +139,38 @@ function loadPage() {
        programEnrollment.$control.oldValue = null;
     });
 
+    // NOTE: As the course details won't be fetched for course id in the CSV
+    // file, this has a potential side-effect of enrolling learners into the courses
+    // which might be marked as closed for reasons other then being "Invite Only".
+    //
+    // This is considered as a reasonable tradeoff at the time of this addition.
+    // Currently, the EnrollmentListView does not support invitation only courses.
+    // This problem does not happen in the Instructor Dashboard because it doesn't 
+    // invoke access checks when calling the enroll method. Modifying the enroll method
+    // is a high-risk change, and it seems that the API will need some changes in
+    // the near future anyway - when the Instructor Dashboard is converted into an
+    // MFE (it could be an excellent opportunity to eliminate many legacy behaviors
+    // there, too).
+    $("#id_bulk_upload_csv").change(function(e) {
+        if (e.target.value) {
+            var force_enrollment = $("#id_force_enrollment");
+            force_enrollment.parent().show();
+            force_enrollment.siblings(".helptext")[0].innerHTML = gettext(
+                "If any of the courses in the CSV file are marked 'Invite Only', " +
+                "this should be enabled for the enrollments to go through in those courses."
+            );
+        }
+    });
+
     if (courseEnrollment.$control.val()) {
         courseEnrollment.$control.trigger("input");
     } else if (programEnrollment.$control.val()) {
         programEnrollment.$control.trigger("input");
     }
+
+    // hide the force_invite_only checkbox by default
+    $("#id_force_enrollment").parent().hide();
+
     $("#learner-management-form").submit(addCheckedLearnersToEnrollBox);
 }
 
diff --git a/enterprise/utils.py b/enterprise/utils.py
index 76b547ade3..e48127d50f 100644
--- a/enterprise/utils.py
+++ b/enterprise/utils.py
@@ -1658,12 +1658,15 @@ def enroll_user(enterprise_customer, user, course_mode, *course_ids, **kwargs):
         user: The user model object who needs to be enrolled in the course
         course_mode: The string representation of the mode with which the enrollment should be created
         *course_ids: An iterable containing any number of course IDs to eventually enroll the user in.
-        kwargs: Should contain enrollment_client if it's already been instantiated and should be passed in.
+        kwargs: Contains optional params such as:
+            - enrollment_client, if it's already been instantiated and should be passed in
+            - force_enrollment, if the course is "Invite Only" and the "force_enrollment" is needed
 
     Returns:
         Boolean: Whether or not enrollment succeeded for all courses specified
     """
     enrollment_client = kwargs.pop('enrollment_client', None)
+    force_enrollment = kwargs.pop('force_enrollment', False)
     if not enrollment_client:
         from enterprise.api_client.lms import EnrollmentApiClient  # pylint: disable=import-outside-toplevel
         enrollment_client = EnrollmentApiClient()
@@ -1678,7 +1681,8 @@ def enroll_user(enterprise_customer, user, course_mode, *course_ids, **kwargs):
                 user.username,
                 course_id,
                 course_mode,
-                enterprise_uuid=str(enterprise_customer_user.enterprise_customer.uuid)
+                enterprise_uuid=str(enterprise_customer_user.enterprise_customer.uuid),
+                force_enrollment=force_enrollment,
             )
         except HttpClientError as exc:
             # Check if user is already enrolled then we should ignore exception
@@ -1956,6 +1960,7 @@ def enroll_users_in_course(
         enrollment_reason=None,
         discount=0.0,
         sales_force_id=None,
+        force_enrollment=False,
 ):
     """
     Enroll existing users in a course, and create a pending enrollment for nonexisting users.
@@ -1969,6 +1974,7 @@ def enroll_users_in_course(
         enrollment_reason (str): A reason for enrollment.
         discount (Decimal): Percentage discount for enrollment.
         sales_force_id (str): Salesforce opportunity id.
+        force_enrollment (bool): Force enrollment into 'Invite Only' courses.
 
     Returns:
         successes: A list of users who were successfully enrolled in the course.
@@ -1985,7 +1991,7 @@ def enroll_users_in_course(
     failures = []
 
     for user in existing_users:
-        succeeded = enroll_user(enterprise_customer, user, course_mode, course_id)
+        succeeded = enroll_user(enterprise_customer, user, course_mode, course_id, force_enrollment=force_enrollment)
         if succeeded:
             successes.append(user)
             if enrollment_requester and enrollment_reason:
diff --git a/test_utils/fake_enrollment_api.py b/test_utils/fake_enrollment_api.py
index 95e7ebfc47..dbf65df3f8 100644
--- a/test_utils/fake_enrollment_api.py
+++ b/test_utils/fake_enrollment_api.py
@@ -150,7 +150,7 @@ def get_course_details(course_id):
         return None
 
 
-def enroll_user_in_course(user, course_id, mode, cohort=None, enterprise_uuid=None):
+def enroll_user_in_course(user, course_id, mode, cohort=None, enterprise_uuid=None, force_enrollment=False):  # pylint: disable=unused-argument
     """
     Fake implementation.
     """
diff --git a/tests/test_admin/test_view.py b/tests/test_admin/test_view.py
index e318aea7ba..480469b1bd 100644
--- a/tests/test_admin/test_view.py
+++ b/tests/test_admin/test_view.py
@@ -820,7 +820,16 @@ def test_post_existing_pending_record_with_another_enterprise_customer(self):
         self._test_post_existing_record_response(response)
         assert PendingEnterpriseCustomerUser.objects.filter(user_email=email).count() == 2
 
-    def _enroll_user_request(self, user, mode, course_id="", notify=True, reason="tests", discount=0.0):
+    def _enroll_user_request(
+        self,
+        user,
+        mode,
+        course_id="",
+        notify=True,
+        reason="tests",
+        discount=0.0,
+        force_enrollment=False,
+    ):
         """
         Perform post request to log in and submit the form to enroll a user.
         """
@@ -844,6 +853,7 @@ def _enroll_user_request(self, user, mode, course_id="", notify=True, reason="te
                 ManageLearnersForm.Fields.NOTIFY: notify,
                 ManageLearnersForm.Fields.REASON: reason,
                 ManageLearnersForm.Fields.DISCOUNT: discount,
+                ManageLearnersForm.Fields.FORCE_ENROLLMENT: force_enrollment,
             })
         return response
 
@@ -902,7 +912,8 @@ def test_post_enroll_user(
             user.username,
             course_id,
             mode,
-            enterprise_uuid=str(self.enterprise_customer.uuid)
+            enterprise_uuid=str(self.enterprise_customer.uuid),
+            force_enrollment=False,
         )
         if enrollment_exists:
             track_enrollment.assert_not_called()
@@ -975,7 +986,8 @@ def _post_multi_enroll(
                     user.username,
                     course_id,
                     mode,
-                    enterprise_uuid=str(self.enterprise_customer.uuid)
+                    enterprise_uuid=str(self.enterprise_customer.uuid),
+                    force_enrollment=False,
                 )
                 track_enrollment.assert_called_with('admin-enrollment', user.id, course_id)
                 self._assert_django_messages(response, {
@@ -1036,20 +1048,71 @@ def test_post_multi_enroll_pending_user(
         """
         Test that a pending learner can be enrolled in multiple courses.
         """
-        self._post_multi_enroll(
+        with mock.patch(
+            'enterprise.models.EnrollmentApiClient.get_course_details',
+            wraps=fake_enrollment_api.get_course_details
+        ):
+            self._post_multi_enroll(
+                enterprise_catalog_client,
+                enrollment_client,
+                course_catalog_client,
+                track_enrollment,
+                False,
+            )
+
+    @mock.patch("enterprise.utils.track_enrollment")
+    @mock.patch("enterprise.models.CourseCatalogApiClient")
+    @mock.patch("enterprise.api_client.lms.EnrollmentApiClient")
+    @mock.patch("enterprise.models.EnterpriseCatalogApiClient")
+    def test_post_enroll_no_course_detail(
+            self,
             enterprise_catalog_client,
             enrollment_client,
             course_catalog_client,
             track_enrollment,
-            False,
+    ):
+        catalog_instance = course_catalog_client.return_value
+        catalog_instance.get_course_run.return_value = {}
+        enrollment_instance = enrollment_client.return_value
+        enrollment_instance.enroll_user_in_course.side_effect = fake_enrollment_api.enroll_user_in_course
+        enrollment_instance.get_course_details.side_effect = fake_enrollment_api.get_course_details
+        enterprise_catalog_instance = enterprise_catalog_client.return_value
+        enterprise_catalog_instance.enterprise_contains_content_items.return_value = True
+
+        user = UserFactory()
+        course_id = "course-v1:HarvardX+CoolScience+2016"
+        mode = "verified"
+        response = self._enroll_user_request(user, mode, course_id=course_id)
+        enrollment_instance.enroll_user_in_course.assert_called_once_with(
+            user.username,
+            course_id,
+            mode,
+            enterprise_uuid=str(self.enterprise_customer.uuid),
+            force_enrollment=False,
         )
+        track_enrollment.assert_called_once_with('admin-enrollment', user.id, course_id)
+        self._assert_django_messages(response, {
+            (messages.SUCCESS, "1 learner was enrolled in {}.".format(course_id)),
+        })
+        all_enterprise_enrollments = EnterpriseCourseEnrollment.objects.all()
+        num_enterprise_enrollments = len(all_enterprise_enrollments)
+        assert num_enterprise_enrollments == 1
+        enrollment = all_enterprise_enrollments[0]
+        assert enrollment.enterprise_customer_user.user == user
+        assert enrollment.course_id == course_id
+        assert enrollment.source is not None
+        assert enrollment.source.slug == EnterpriseEnrollmentSource.MANUAL
+        num_messages = len(mail.outbox)
+        assert num_messages == 0
 
     @mock.patch("enterprise.utils.track_enrollment")
     @mock.patch("enterprise.models.CourseCatalogApiClient")
     @mock.patch("enterprise.api_client.lms.EnrollmentApiClient")
     @mock.patch("enterprise.models.EnterpriseCatalogApiClient")
-    def test_post_enroll_no_course_detail(
+    @ddt.data(True, False)
+    def test_post_enroll_force_enrollment(
             self,
+            force_enrollment,
             enterprise_catalog_client,
             enrollment_client,
             course_catalog_client,
@@ -1066,12 +1129,13 @@ def test_post_enroll_no_course_detail(
         user = UserFactory()
         course_id = "course-v1:HarvardX+CoolScience+2016"
         mode = "verified"
-        response = self._enroll_user_request(user, mode, course_id=course_id)
+        response = self._enroll_user_request(user, mode, course_id=course_id, force_enrollment=force_enrollment)
         enrollment_instance.enroll_user_in_course.assert_called_once_with(
             user.username,
             course_id,
             mode,
-            enterprise_uuid=str(self.enterprise_customer.uuid)
+            enterprise_uuid=str(self.enterprise_customer.uuid),
+            force_enrollment=force_enrollment,
         )
         track_enrollment.assert_called_once_with('admin-enrollment', user.id, course_id)
         self._assert_django_messages(response, {
@@ -1085,8 +1149,6 @@ def test_post_enroll_no_course_detail(
         assert enrollment.course_id == course_id
         assert enrollment.source is not None
         assert enrollment.source.slug == EnterpriseEnrollmentSource.MANUAL
-        num_messages = len(mail.outbox)
-        assert num_messages == 0
 
     @mock.patch("enterprise.utils.track_enrollment")
     @mock.patch("enterprise.models.CourseCatalogApiClient")
@@ -1136,7 +1198,8 @@ def test_post_enroll_course_when_enrollment_closed(
             user.username,
             course_id,
             mode,
-            enterprise_uuid=str(self.enterprise_customer.uuid)
+            enterprise_uuid=str(self.enterprise_customer.uuid),
+            force_enrollment=False,
         )
 
     @mock.patch("enterprise.utils.track_enrollment")
@@ -1170,7 +1233,8 @@ def test_post_enroll_course_when_enrollment_closed_mode_changed(
             user.username,
             course_id,
             mode,
-            enterprise_uuid=str(self.enterprise_customer.uuid)
+            enterprise_uuid=str(self.enterprise_customer.uuid),
+            force_enrollment=False,
         )
         track_enrollment.assert_not_called()
         self._assert_django_messages(response, {
@@ -1211,7 +1275,8 @@ def test_post_enroll_course_when_enrollment_closed_no_sce_exists(
             user.username,
             course_id,
             mode,
-            enterprise_uuid=str(self.enterprise_customer.uuid)
+            enterprise_uuid=str(self.enterprise_customer.uuid),
+            force_enrollment=False,
         )
         track_enrollment.assert_not_called()
         self._assert_django_messages(response, {
@@ -1256,7 +1321,8 @@ def test_post_enroll_with_missing_course_start_date(
             user.username,
             course_id,
             mode,
-            enterprise_uuid=str(self.enterprise_customer.uuid)
+            enterprise_uuid=str(self.enterprise_customer.uuid),
+            force_enrollment=False,
         )
         track_enrollment.assert_called_once_with('admin-enrollment', user.id, course_id)
         self._assert_django_messages(response, {
@@ -1570,6 +1636,7 @@ def test_post_create_course_enrollments(
                 enrollment_requester=ANY,
                 enterprise_customer=ANY,
                 sales_force_id=ANY,
+                force_enrollment=ANY,
             )
             enroll_users_in_course_mock.assert_any_call(
                 course_id=second_course_id,
@@ -1580,6 +1647,7 @@ def test_post_create_course_enrollments(
                 enrollment_requester=ANY,
                 enterprise_customer=ANY,
                 sales_force_id=ANY,
+                force_enrollment=ANY,
             )
         else:
             enroll_users_in_course_mock.assert_not_called()
@@ -1664,8 +1732,10 @@ def test_post_successful_test(self):
     @mock.patch("enterprise.models.CourseCatalogApiClient")
     @mock.patch("enterprise.api_client.lms.EnrollmentApiClient")
     @mock.patch("enterprise.models.EnterpriseCatalogApiClient")
+    @mock.patch("enterprise.models.CourseEnrollmentAllowed")
     def test_post_link_and_enroll(
             self,
+            mock_cea,
             enterprise_catalog_client,
             enrollment_client,
             course_catalog_client,
@@ -1698,13 +1768,18 @@ def test_post_link_and_enroll(
         course_id = "course-v1:EnterpriseX+Training+2017"
         course_mode = "professional"
 
-        response = self._perform_request(columns, data, course=course_id, course_mode=course_mode)
+        with mock.patch(
+            'enterprise.models.EnrollmentApiClient.get_course_details',
+            wraps=fake_enrollment_api.get_course_details,
+        ):
+            response = self._perform_request(columns, data, course=course_id, course_mode=course_mode)
 
         enrollment_instance.enroll_user_in_course.assert_called_once_with(
             user.username,
             course_id,
             course_mode,
-            enterprise_uuid=str(self.enterprise_customer.uuid)
+            enterprise_uuid=str(self.enterprise_customer.uuid),
+            force_enrollment=False,
         )
         track_enrollment.assert_called_once_with('admin-enrollment', user.id, course_id)
         pending_user_message = (
@@ -1723,13 +1798,16 @@ def test_post_link_and_enroll(
         assert pending_enrollment.sales_force_id == sales_force_id
         num_messages = len(mail.outbox)
         assert num_messages == 2
+        mock_cea.objects.update_or_create.assert_called_once()
 
     @mock.patch("enterprise.utils.track_enrollment")
     @mock.patch("enterprise.models.CourseCatalogApiClient")
     @mock.patch("enterprise.api_client.lms.EnrollmentApiClient")
     @mock.patch("enterprise.models.EnterpriseCatalogApiClient")
+    @mock.patch("enterprise.models.CourseEnrollmentAllowed")
     def test_post_link_and_enroll_no_course_details(
             self,
+            mock_cea,
             enterprise_catalog_client,
             enrollment_client,
             course_catalog_client,
@@ -1754,13 +1832,18 @@ def test_post_link_and_enroll_no_course_details(
         course_id = "course-v1:EnterpriseX+Training+2017"
         course_mode = "professional"
 
-        response = self._perform_request(columns, data, course=course_id, course_mode=course_mode)
+        with mock.patch(
+            'enterprise.models.EnrollmentApiClient.get_course_details',
+            wraps=fake_enrollment_api.get_course_details,
+        ):
+            response = self._perform_request(columns, data, course=course_id, course_mode=course_mode)
 
         enrollment_instance.enroll_user_in_course.assert_called_once_with(
             user.username,
             course_id,
             course_mode,
-            enterprise_uuid=str(self.enterprise_customer.uuid)
+            enterprise_uuid=str(self.enterprise_customer.uuid),
+            force_enrollment=False,
         )
         track_enrollment.assert_called_once_with('admin-enrollment', user.id, course_id)
         pending_user_message = (
@@ -1776,12 +1859,15 @@ def test_post_link_and_enroll_no_course_details(
         assert PendingEnterpriseCustomerUser.objects.all()[0].pendingenrollment_set.all()[0].course_id == course_id
         num_messages = len(mail.outbox)
         assert num_messages == 0
+        mock_cea.objects.update_or_create.assert_called_once()
 
     @mock.patch("enterprise.utils.track_enrollment")
     @mock.patch("enterprise.api_client.lms.EnrollmentApiClient")
     @mock.patch("enterprise.models.EnterpriseCatalogApiClient")
+    @mock.patch("enterprise.models.CourseEnrollmentAllowed")
     def test_post_link_and_enroll_no_notification(
             self,
+            mock_cea,
             enterprise_catalog_client,
             enrollment_client,
             track_enrollment,
@@ -1803,13 +1889,18 @@ def test_post_link_and_enroll_no_notification(
         course_id = "course-v1:EnterpriseX+Training+2017"
         course_mode = "professional"
 
-        response = self._perform_request(columns, data, course=course_id, course_mode=course_mode, notify=False)
+        with mock.patch(
+            'enterprise.models.EnrollmentApiClient.get_course_details',
+            wraps=fake_enrollment_api.get_course_details,
+        ):
+            response = self._perform_request(columns, data, course=course_id, course_mode=course_mode, notify=False)
 
         enrollment_instance.enroll_user_in_course.assert_called_once_with(
             user.username,
             course_id,
             course_mode,
-            enterprise_uuid=str(self.enterprise_customer.uuid)
+            enterprise_uuid=str(self.enterprise_customer.uuid),
+            force_enrollment=False,
         )
         track_enrollment.assert_called_once_with('admin-enrollment', user.id, course_id)
         pending_user_message = (
@@ -1824,6 +1915,7 @@ def test_post_link_and_enroll_no_notification(
         assert PendingEnterpriseCustomerUser.objects.all()[0].pendingenrollment_set.all()[0].course_id == course_id
         num_messages = len(mail.outbox)
         assert num_messages == 0
+        mock_cea.objects.update_or_create.assert_called_once()
 
 
 @mark.django_db
diff --git a/tests/test_enterprise/api/test_views.py b/tests/test_enterprise/api/test_views.py
index af831d3302..2be0f0b281 100644
--- a/tests/test_enterprise/api/test_views.py
+++ b/tests/test_enterprise/api/test_views.py
@@ -67,7 +67,7 @@
 )
 from test_utils.decorators import mock_api_response
 from test_utils.factories import FAKER, PendingEnterpriseCustomerUserFactory
-from test_utils.fake_enterprise_api import get_default_branding_object
+from test_utils.fake_enrollment_api import get_course_details, get_default_branding_object
 
 fake = Faker()
 
@@ -2600,6 +2600,7 @@ def test_enterprise_customer_course_enrollments_detail_success(
             True,
             enable_autocohorting=True
         )
+        mock_enrollment_client.return_value.get_course_details = get_course_details
 
         # Make the call!
         response = self.client.post(
@@ -2797,7 +2798,8 @@ def test_enterprise_customer_course_enrollments_detail_multiple(
             get_course_enrollment=mock.Mock(
                 side_effect=[None, {'is_active': True, 'mode': VERIFIED_SUBSCRIPTION_COURSE_MODE}]
             ),
-            enroll_user_in_course=mock.Mock()
+            enroll_user_in_course=mock.Mock(),
+            get_course_details=get_course_details
         )
 
         # Set up catalog_contains_course response.
@@ -3359,6 +3361,7 @@ class TestBulkEnrollment(BaseTestEnterpriseAPIViews):
             'expected_response': {'non_field_errors': ['Must include the "licenses_info" parameter in request.']},
             'expected_num_pending_licenses': 0,
             'expected_events': None,
+            'expected_cea': 0,
         },
         {
             'body': {
@@ -3370,6 +3373,7 @@ class TestBulkEnrollment(BaseTestEnterpriseAPIViews):
             },
             'expected_num_pending_licenses': 0,
             'expected_events': None,
+            'expected_cea': 0,
         },
         {
             'body': {
@@ -3381,6 +3385,7 @@ class TestBulkEnrollment(BaseTestEnterpriseAPIViews):
             },
             'expected_num_pending_licenses': 0,
             'expected_events': None,
+            'expected_cea': 0,
         },
         {
             'body': {
@@ -3396,6 +3401,7 @@ class TestBulkEnrollment(BaseTestEnterpriseAPIViews):
             },
             'expected_num_pending_licenses': 0,
             'expected_events': None,
+            'expected_cea': 0,
         },
         # Single learner, single course success
         {
@@ -3419,6 +3425,7 @@ class TestBulkEnrollment(BaseTestEnterpriseAPIViews):
             },
             'expected_num_pending_licenses': 1,
             'expected_events': [mock.call(PATHWAY_CUSTOMER_ADMIN_ENROLLMENT, 1, 'course-v1:edX+DemoX+Demo_Course')],
+            'expected_cea': 0,
         },
         # Multi-learner, single course success
         {
@@ -3459,6 +3466,7 @@ class TestBulkEnrollment(BaseTestEnterpriseAPIViews):
             'expected_events': [
                 mock.call(PATHWAY_CUSTOMER_ADMIN_ENROLLMENT, 1, 'course-v1:edX+DemoX+Demo_Course'),
             ],
+            'expected_cea': 0,
         },
         # Multi-learner, multi-course success
         {
@@ -3476,12 +3484,12 @@ class TestBulkEnrollment(BaseTestEnterpriseAPIViews):
                     },
                     {
                         'email': 'abc@test.com',
-                        'course_run_key': 'course-v2:edX+DemoX+Second_Demo_Course',
+                        'course_run_key': 'course-v1:EnterpriseX+Training+2017',
                         'license_uuid': '5a88bdcade7c4ecb838f8111b68e18ac'
                     },
                     {
                         'email': 'xyz@test.com',
-                        'course_run_key': 'course-v2:edX+DemoX+Second_Demo_Course',
+                        'course_run_key': 'course-v1:EnterpriseX+Training+2017',
                         'license_uuid': '2c58acdade7c4ede838f7111b42e18ac'
                     },
                 ]
@@ -3504,13 +3512,13 @@ class TestBulkEnrollment(BaseTestEnterpriseAPIViews):
                     },
                     {
                         'email': 'abc@test.com',
-                        'course_run_key': 'course-v2:edX+DemoX+Second_Demo_Course',
+                        'course_run_key': 'course-v1:EnterpriseX+Training+2017',
                         'created': True,
                         'activation_link': None,
                     },
                     {
                         'email': 'xyz@test.com',
-                        'course_run_key': 'course-v2:edX+DemoX+Second_Demo_Course',
+                        'course_run_key': 'course-v1:EnterpriseX+Training+2017',
                         'created': True,
                         'activation_link': None,
                     }
@@ -3520,16 +3528,19 @@ class TestBulkEnrollment(BaseTestEnterpriseAPIViews):
             'expected_num_pending_licenses': 4,
             'expected_events': [
                 mock.call(PATHWAY_CUSTOMER_ADMIN_ENROLLMENT, 1, 'course-v1:edX+DemoX+Demo_Course'),
-                mock.call(PATHWAY_CUSTOMER_ADMIN_ENROLLMENT, 1, 'course-v2:edX+DemoX+Second_Demo_Course')
+                mock.call(PATHWAY_CUSTOMER_ADMIN_ENROLLMENT, 1, 'course-v1:EnterpriseX+Training+2017')
             ],
+            'expected_cea': 2
         },
     )
     @ddt.unpack
     @mock.patch('enterprise.api.v1.views.get_best_mode_from_course_key')
     @mock.patch('enterprise.api.v1.views.track_enrollment')
     @mock.patch("enterprise.models.EnterpriseCustomer.notify_enrolled_learners")
+    @mock.patch("enterprise.models.CourseEnrollmentAllowed")
     def test_bulk_enrollment_in_bulk_courses_pending_licenses(
         self,
+        mock_cea,
         mock_notify_task,
         mock_track_enroll,
         mock_get_course_mode,
@@ -3538,6 +3549,7 @@ def test_bulk_enrollment_in_bulk_courses_pending_licenses(
         expected_response,
         expected_num_pending_licenses,
         expected_events,
+        expected_cea,
     ):
         """
         Tests the bulk enrollment endpoint at enroll_learners_in_courses.
@@ -3554,11 +3566,17 @@ def test_bulk_enrollment_in_bulk_courses_pending_licenses(
         mock_get_course_mode.return_value = VERIFIED_SUBSCRIPTION_COURSE_MODE
 
         self.assertEqual(len(PendingEnrollment.objects.all()), 0)
-        response = self.client.post(
-            settings.TEST_SERVER + ENTERPRISE_CUSTOMER_BULK_ENROLL_LEARNERS_IN_COURSES_ENDPOINT,
-            data=json.dumps(body),
-            content_type='application/json',
-        )
+
+        with mock.patch(
+            "enterprise.models.EnrollmentApiClient.get_course_details",
+            wraps=get_course_details
+        ):
+            response = self.client.post(
+                settings.TEST_SERVER + ENTERPRISE_CUSTOMER_BULK_ENROLL_LEARNERS_IN_COURSES_ENDPOINT,
+                data=json.dumps(body),
+                content_type='application/json',
+            )
+
         self.assertEqual(response.status_code, expected_code)
         if expected_response:
             response_json = response.json()
@@ -3573,6 +3591,7 @@ def test_bulk_enrollment_in_bulk_courses_pending_licenses(
         else:
             mock_track_enroll.assert_not_called()
 
+        self.assertEqual(mock_cea.objects.update_or_create.call_count, expected_cea)
         # no notifications to be sent unless 'notify' specifically asked for in payload
         mock_notify_task.assert_not_called()
 
@@ -3621,13 +3640,13 @@ def test_bulk_enrollment_in_bulk_courses_pending_licenses(
                     },
                     {
                         'email': 'abc@test.com',
-                        'course_run_key': 'course-v2:edX+DemoX+Second_Demo_Course',
+                        'course_run_key': 'course-v1:HarvardX+CoolScience+2016',
                         'created': True,
                         'activation_link': None,
                     },
                     {
                         'email': 'xyz@test.com',
-                        'course_run_key': 'course-v2:edX+DemoX+Second_Demo_Course',
+                        'course_run_key': 'course-v1:HarvardX+CoolScience+2016',
                         'created': True,
                         'activation_link': None,
                     }
@@ -3637,7 +3656,7 @@ def test_bulk_enrollment_in_bulk_courses_pending_licenses(
             'expected_num_pending_licenses': 4,
             'expected_events': [
                 mock.call(PATHWAY_CUSTOMER_ADMIN_ENROLLMENT, 1, 'course-v1:edX+DemoX+Demo_Course'),
-                mock.call(PATHWAY_CUSTOMER_ADMIN_ENROLLMENT, 1, 'course-v2:edX+DemoX+Second_Demo_Course')
+                mock.call(PATHWAY_CUSTOMER_ADMIN_ENROLLMENT, 1, 'course-v1:HarvardX+CoolScience+2016')
             ],
         },
     )
@@ -3672,13 +3691,14 @@ def test_bulk_enrollment_with_notification(
 
         self.assertEqual(len(PendingEnrollment.objects.all()), 0)
 
-        response = self.client.post(
-            settings.TEST_SERVER + ENTERPRISE_CUSTOMER_BULK_ENROLL_LEARNERS_IN_COURSES_ENDPOINT,
-            data=json.dumps(body),
-            content_type='application/json',
-        )
-        self.assertEqual(response.status_code, expected_code)
+        with mock.patch("enterprise.models.EnrollmentApiClient.get_course_details", wraps=get_course_details):
+            response = self.client.post(
+                settings.TEST_SERVER + ENTERPRISE_CUSTOMER_BULK_ENROLL_LEARNERS_IN_COURSES_ENDPOINT,
+                data=json.dumps(body),
+                content_type='application/json',
+            )
 
+        self.assertEqual(response.status_code, expected_code)
         response_json = response.json()
         self.assertEqual(expected_response, response_json)
         self.assertEqual(len(PendingEnrollment.objects.all()), expected_num_pending_licenses)
diff --git a/tests/test_enterprise/api_client/test_lms.py b/tests/test_enterprise/api_client/test_lms.py
index 6ae3715d9e..f73cc75286 100644
--- a/tests/test_enterprise/api_client/test_lms.py
+++ b/tests/test_enterprise/api_client/test_lms.py
@@ -94,7 +94,8 @@ def test_enroll_user_in_course():
         mode=mode,
         cohort=cohort,
         is_active=True,
-        enterprise_uuid='None'
+        enterprise_uuid='None',
+        force_enrollment=False
     )
     responses.add(
         responses.POST,
diff --git a/tests/test_enterprise/test_utils.py b/tests/test_enterprise/test_utils.py
index ba3b5bb0b6..884627e886 100644
--- a/tests/test_enterprise/test_utils.py
+++ b/tests/test_enterprise/test_utils.py
@@ -21,6 +21,7 @@
     serialize_notification_content,
 )
 from test_utils import FAKE_UUIDS, TEST_PASSWORD, TEST_USERNAME, factories
+from test_utils.fake_enrollment_api import get_course_details
 
 LMS_BASE_URL = 'https://lms.base.url'
 
@@ -243,11 +244,12 @@ def test_enroll_pending_licensed_users_in_courses_succeeds(self):
         )
         licensed_users_info = [{
             'email': 'pending-user-email@example.com',
-            'course_run_key': 'course-key-v1',
+            'course_run_key': 'course-v1:edX+DemoX+Demo_Course',
             'course_mode': 'verified',
             'license_uuid': '5b77bdbade7b4fcb838f8111b68e18ae'
         }]
-        result = enroll_licensed_users_in_courses(ent_customer, licensed_users_info)
+        with mock.patch("enterprise.models.EnrollmentApiClient.get_course_details", wraps=get_course_details):
+            result = enroll_licensed_users_in_courses(ent_customer, licensed_users_info)
 
         self.assertEqual(result['pending'][0]['email'], 'pending-user-email@example.com')
         self.assertFalse(result['successes'])

From 6e74b8b545e0bb6c7fec44db19bb715f132a3af2 Mon Sep 17 00:00:00 2001
From: Arunmozhi <arunmozhi@opencraft.com>
Date: Fri, 26 May 2023 22:28:27 +0530
Subject: [PATCH 2/9] fix: add missing CEA definition and fix imports in tests

---
 enterprise/models.py                    | 1 +
 tests/test_enterprise/api/test_views.py | 7 ++++---
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/enterprise/models.py b/enterprise/models.py
index f2e0615f36..07d90cf7ec 100644
--- a/enterprise/models.py
+++ b/enterprise/models.py
@@ -78,6 +78,7 @@
     from common.djangoapps.student.models import CourseEnrollment, CourseEnrollmentAllowed
 except ImportError:
     CourseEnrollment = None
+    CourseEnrollmentAllowed = None
 
 try:
     from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
diff --git a/tests/test_enterprise/api/test_views.py b/tests/test_enterprise/api/test_views.py
index 2be0f0b281..4fdb9f4b4d 100644
--- a/tests/test_enterprise/api/test_views.py
+++ b/tests/test_enterprise/api/test_views.py
@@ -67,7 +67,8 @@
 )
 from test_utils.decorators import mock_api_response
 from test_utils.factories import FAKER, PendingEnterpriseCustomerUserFactory
-from test_utils.fake_enrollment_api import get_course_details, get_default_branding_object
+from test_utils.fake_enrollment_api import get_course_details
+from test_utils.fake_enterprise_api import get_default_branding_object
 
 fake = Faker()
 
@@ -3612,12 +3613,12 @@ def test_bulk_enrollment_in_bulk_courses_pending_licenses(
                     },
                     {
                         'email': 'abc@test.com',
-                        'course_run_key': 'course-v2:edX+DemoX+Second_Demo_Course',
+                        'course_run_key': 'course-v1:HarvardX+CoolScience+2016',
                         'license_uuid': '5a88bdcade7c4ecb838f8111b68e18ac'
                     },
                     {
                         'email': 'xyz@test.com',
-                        'course_run_key': 'course-v2:edX+DemoX+Second_Demo_Course',
+                        'course_run_key': 'course-v1:HarvardX+CoolScience+2016',
                         'license_uuid': '2c58acdade7c4ede838f7111b42e18ac'
                     },
                 ]

From 0c23adf9f86928af570cac789996b482a912ece0 Mon Sep 17 00:00:00 2001
From: Arunmozhi <arun@arunmozhi.in>
Date: Sun, 28 May 2023 16:23:32 +0530
Subject: [PATCH 3/9] fix: rollback all changes in admin forms

---
 enterprise/admin/forms.py | 261 +++++++++++++-------------------------
 1 file changed, 91 insertions(+), 170 deletions(-)

diff --git a/enterprise/admin/forms.py b/enterprise/admin/forms.py
index b1a4c9a2a1..fddb2408ee 100644
--- a/enterprise/admin/forms.py
+++ b/enterprise/admin/forms.py
@@ -19,10 +19,7 @@
 from django.utils.translation import gettext as _
 
 from enterprise import utils
-from enterprise.admin.utils import (
-    email_or_username__to__email,
-    split_usernames_and_emails,
-)
+from enterprise.admin.utils import email_or_username__to__email, split_usernames_and_emails
 from enterprise.admin.widgets import SubmitInput
 from enterprise.models import (
     AdminNotification,
@@ -37,9 +34,7 @@
 from enterprise.utils import ValidationMessages, validate_email_to_link
 
 try:
-    from common.djangoapps.third_party_auth.models import (
-        SAMLProviderConfig as saml_provider_configuration,
-    )
+    from common.djangoapps.third_party_auth.models import SAMLProviderConfig as saml_provider_configuration
 except ImportError:
     saml_provider_configuration = None
 
@@ -51,11 +46,10 @@ class ManageLearnersForm(forms.Form):
     """
     Form to manage learner additions.
     """
-
     email_or_username = forms.CharField(
-        label=_("To add a single learner, enter an email address or username."),
-        required=False,
-    )
+        label=_(
+            "To add a single learner, enter an email address or username."),
+        required=False)
     bulk_upload_csv = forms.FileField(
         label=_(
             "To add multiple learners, upload a .csv file that contains a "
@@ -67,25 +61,15 @@ class ManageLearnersForm(forms.Form):
             "by the heading 'email' in the first row. Optionally, the .csv file may contain "
             "a column of course run keys, indicated by the heading 'course_id' in the first row, to "
             "enroll learners in multiple courses."
-        ),
+        )
     )
     course = forms.CharField(
-        label=_("Enroll these learners in this course"),
-        required=False,
+        label=_("Enroll these learners in this course"), required=False,
         help_text=_("To enroll learners in a course, enter a course ID."),
     )
-    force_enrollment = forms.BooleanField(
-        label=_("Force Enrollment"),
-        help_text=_(
-            "The selected course is 'Invite Only'. Only staff can enroll learners to this course."
-        ),
-        required=False,
-    )
     course_mode = forms.ChoiceField(
-        label=_("Course enrollment track"),
-        required=False,
-        choices=BLANK_CHOICE_DASH
-        + [
+        label=_("Course enrollment track"), required=False,
+        choices=BLANK_CHOICE_DASH + [
             ("audit", _("Audit")),
             ("verified", _("Verified")),
             ("professional", _("Professional Education")),
@@ -95,27 +79,22 @@ class ManageLearnersForm(forms.Form):
         ],
     )
     reason = forms.CharField(label=_("Reason for manual enrollment"), required=False)
-    sales_force_id = forms.CharField(
-        label=_("Salesforce Opportunity ID"), required=False
-    )
+    sales_force_id = forms.CharField(label=_("Salesforce Opportunity ID"), required=False)
     discount = forms.DecimalField(
         label=_("Discount percentage for manual enrollment"),
         help_text=_("Discount percentage should be from 0 to 100"),
         required=True,
         decimal_places=5,
-        initial=0.0,
+        initial=0.0
     )
 
     class NotificationTypes:
         """
         Namespace class for notification types
         """
-
-        BY_EMAIL = "by_email"
-        NO_NOTIFICATION = "do_not_notify"
-        DEFAULT = getattr(
-            settings, "DEFAULT_ENTERPRISE_NOTIFICATION_MECHANISM", BY_EMAIL
-        )
+        BY_EMAIL = 'by_email'
+        NO_NOTIFICATION = 'do_not_notify'
+        DEFAULT = getattr(settings, 'DEFAULT_ENTERPRISE_NOTIFICATION_MECHANISM', BY_EMAIL)
 
     notify_on_enrollment = forms.ChoiceField(
         label=_("Notify learners of enrollment"),
@@ -131,7 +110,6 @@ class Modes:
         """
         Namespace class for form modes.
         """
-
         MODE_SINGULAR = "singular"
         MODE_BULK = "bulk"
 
@@ -139,7 +117,6 @@ class Fields:
         """
         Namespace class for field names.
         """
-
         GENERAL_ERRORS = forms.forms.NON_FIELD_ERRORS
 
         EMAIL_OR_USERNAME = "email_or_username"
@@ -151,13 +128,11 @@ class Fields:
         REASON = "reason"
         SALES_FORCE_ID = "sales_force_id"
         DISCOUNT = "discount"
-        FORCE_ENROLLMENT = "force_enrollment"
 
     class CsvColumns:
         """
         Namespace class for CSV column names.
         """
-
         EMAIL = "email"
         COURSE_ID = "course_id"
 
@@ -170,9 +145,9 @@ def __init__(self, *args, **kwargs):
             user (django.contrib.auth.models.User): current user
             enterprise_customer (enterprise.models.EnterpriseCustomer): current customer
         """
-        user = kwargs.pop("user", None)
+        user = kwargs.pop('user', None)
         self._user = user
-        self._enterprise_customer = kwargs.pop("enterprise_customer", None)
+        self._enterprise_customer = kwargs.pop('enterprise_customer', None)
         super().__init__(*args, **kwargs)
 
     def clean_email_or_username(self):
@@ -227,12 +202,8 @@ def clean_course(self):
         course_id = self.cleaned_data[self.Fields.COURSE].strip()
         if not course_id:
             return None
-        enterprise_customer = utils.get_enterprise_customer(
-            self._enterprise_customer.uuid
-        )
-        course_details = utils.validate_course_exists_for_enterprise(
-            enterprise_customer, course_id
-        )
+        enterprise_customer = utils.get_enterprise_customer(self._enterprise_customer.uuid)
+        course_details = utils.validate_course_exists_for_enterprise(enterprise_customer, course_id)
         return course_details
 
     def clean_reason(self):
@@ -289,15 +260,11 @@ def _validate_bulk_upload(self):
         bulk_upload_csv = self.cleaned_data.get(self.Fields.BULK_UPLOAD)
         if bulk_upload_csv:
             bulk_csv_contents = bulk_upload_csv.read()
-            csv_column_names = (
-                bulk_csv_contents.decode("utf-8").split("\n", 1)[0].split(",")
-            )
+            csv_column_names = bulk_csv_contents.decode('utf-8').split('\n', 1)[0].split(',')
             if self.CsvColumns.COURSE_ID in csv_column_names:
                 # course id column exists in csv, so validate conditionally required fields
                 if self.cleaned_data.get(self.Fields.COURSE):
-                    raise ValidationError(
-                        ValidationMessages.BOTH_COURSE_FIELDS_SPECIFIED
-                    )
+                    raise ValidationError(ValidationMessages.BOTH_COURSE_FIELDS_SPECIFIED)
                 if not self.cleaned_data.get(self.Fields.COURSE_MODE):
                     raise ValidationError(ValidationMessages.COURSE_WITHOUT_COURSE_MODE)
                 if not self.cleaned_data.get(self.Fields.REASON):
@@ -315,12 +282,10 @@ def _validate_course(self):
                 raise ValidationError(ValidationMessages.COURSE_WITHOUT_COURSE_MODE)
             valid_course_modes = course_details["course_modes"]
             if all(course_mode != mode["slug"] for mode in valid_course_modes):
-                error = ValidationError(
-                    ValidationMessages.COURSE_MODE_INVALID_FOR_COURSE.format(
-                        course_mode=course_mode,
-                        course_id=course_details["course_id"],
-                    )
-                )
+                error = ValidationError(ValidationMessages.COURSE_MODE_INVALID_FOR_COURSE.format(
+                    course_mode=course_mode,
+                    course_id=course_details["course_id"],
+                ))
                 raise ValidationError({self.Fields.COURSE_MODE: error})
 
     def _validate_reason(self):
@@ -339,11 +304,10 @@ class ManageLearnersDataSharingConsentForm(forms.Form):
     """
     Form to request DSC from a learner.
     """
-
     email_or_username = forms.CharField(
         label=_("Email/Username"),
         help_text=_("Enter an email address or username."),
-        required=True,
+        required=True
     )
     course = forms.CharField(
         label=_("Course"),
@@ -355,7 +319,6 @@ class Fields:
         """
         Namespace class for field names.
         """
-
         EMAIL_OR_USERNAME = "email_or_username"
         COURSE = "course"
 
@@ -366,7 +329,7 @@ def __init__(self, *args, **kwargs):
         Arguments:
             enterprise_customer (enterprise.models.EnterpriseCustomer): current customer
         """
-        self._enterprise_customer = kwargs.pop("enterprise_customer", None)
+        self._enterprise_customer = kwargs.pop('enterprise_customer', None)
         super().__init__(*args, **kwargs)
 
     def clean_course(self):
@@ -376,9 +339,7 @@ def clean_course(self):
         course_id = self.cleaned_data[self.Fields.COURSE].strip()
         if not course_id:
             return None
-        enterprise_customer = utils.get_enterprise_customer(
-            self._enterprise_customer.uuid
-        )
+        enterprise_customer = utils.get_enterprise_customer(self._enterprise_customer.uuid)
         utils.validate_course_exists_for_enterprise(enterprise_customer, course_id)
         return course_id
 
@@ -401,9 +362,7 @@ def is_course_in_catalog(self, course_id):
         """
         Check whether course exists in enterprise customer catalog.
         """
-        enterprise_customer = utils.get_enterprise_customer(
-            self._enterprise_customer.uuid
-        )
+        enterprise_customer = utils.get_enterprise_customer(self._enterprise_customer.uuid)
         return enterprise_customer.catalog_contains_course(course_id)
 
     def is_user_linked(self, email):
@@ -411,16 +370,13 @@ def is_user_linked(self, email):
         Check whether user is linked to the enterprise customer or not.
         """
         user = User.objects.get(email=email)
-        return utils.get_enterprise_customer_user(
-            user.id, self._enterprise_customer.uuid
-        )
+        return utils.get_enterprise_customer_user(user.id, self._enterprise_customer.uuid)
 
 
 class EnterpriseCustomerAdminForm(forms.ModelForm):
     """
     Alternate form for the EnterpriseCustomer admin page.
     """
-
     class Meta:
         model = EnterpriseCustomer
         fields = (
@@ -458,19 +414,14 @@ class Meta:
 
 class EnterpriseCustomerCatalogAdminForm(forms.ModelForm):
     """
-    form for EnterpriseCustomerCatalogAdmin class.
+        form for EnterpriseCustomerCatalogAdmin class.
     """
-
     class Meta:
         model = EnterpriseCustomerCatalog
         fields = "__all__"
 
-    preview_button = forms.Field(
-        required=False,
-        label="Actions",
-        widget=SubmitInput(attrs={"value": _("Preview")}),
-        help_text=_("Hold Ctrl when clicking on button to open Preview in new tab"),
-    )
+    preview_button = forms.Field(required=False, label='Actions', widget=SubmitInput(attrs={'value': _('Preview')}),
+                                 help_text=_("Hold Ctrl when clicking on button to open Preview in new tab"))
 
     @staticmethod
     def get_catalog_preview_uuid(post_data):
@@ -480,22 +431,16 @@ def get_catalog_preview_uuid(post_data):
 
         e.g: 'enterprise_customer_catalogs-0-preview_button'
         """
-        preview_button_expression = re.compile(
-            r"enterprise_customer_catalogs-\d+-preview_button"
-        )
-        clicked_button_index_expression = re.compile(r"-(.+?)-")
+        preview_button_expression = re.compile(r'enterprise_customer_catalogs-\d+-preview_button')
+        clicked_button_index_expression = re.compile(r'-(.+?)-')
         count = 0
         preview_button_index = None
         for key, _ in post_data.items():
             if preview_button_expression.match(key):
                 count += 1
-                preview_button_index = clicked_button_index_expression.search(
-                    key
-                ).group(1)
+                preview_button_index = clicked_button_index_expression.search(key).group(1)
         if count == 1:
-            return post_data.get(
-                "enterprise_customer_catalogs-" + preview_button_index + "-uuid"
-            )
+            return post_data.get('enterprise_customer_catalogs-' + preview_button_index + '-uuid')
         return None
 
 
@@ -506,7 +451,6 @@ class EnterpriseCustomerIdentityProviderAdminForm(forms.ModelForm):
     This form fetches identity providers from lms third_party_auth app.
     If third_party_auth app is not avilable it displays provider_id as a CharField.
     """
-
     class Meta:
         model = EnterpriseCustomerIdentityProvider
         fields = "__all__"
@@ -519,34 +463,28 @@ def __init__(self, *args, **kwargs):
         """
         super().__init__(*args, **kwargs)
         idp_choices = utils.get_idp_choices()
-        help_text = ""
+        help_text = ''
         if saml_provider_configuration:
             provider_id = self.instance.provider_id
-            url = reverse(
-                "admin:{}_{}_add".format(
-                    saml_provider_configuration._meta.app_label,
-                    saml_provider_configuration._meta.model_name,
-                )
-            )
+            url = reverse('admin:{}_{}_add'.format(
+                saml_provider_configuration._meta.app_label,
+                saml_provider_configuration._meta.model_name))
             if provider_id:
                 identity_provider = utils.get_identity_provider(provider_id)
                 if identity_provider:
-                    update_url = url + "?source={}".format(identity_provider.pk)
-                    help_text = '<p><a href="{update_url}" target="_blank">View "{identity_provider}" details</a><p>'.format(
-                        update_url=update_url, identity_provider=identity_provider.name
-                    )
+                    update_url = url + '?source={}'.format(identity_provider.pk)
+                    help_text = '<p><a href="{update_url}" target="_blank">View "{identity_provider}" details</a><p>'.\
+                        format(update_url=update_url, identity_provider=identity_provider.name)
                 else:
                     help_text += '<p style="margin-top:-5px;"> Make sure you have added a valid provider_id.</p>'
             else:
-                help_text += (
-                    '<p style="margin-top:-5px;"><a target="_blank" href={add_url}>'
-                    "Create a new identity provider</a></p>".format(add_url=url)
-                )
+                help_text += '<p style="margin-top:-5px;"><a target="_blank" href={add_url}>' \
+                             'Create a new identity provider</a></p>'.format(add_url=url)
 
         if idp_choices is not None:
-            self.fields["provider_id"] = forms.TypedChoiceField(
+            self.fields['provider_id'] = forms.TypedChoiceField(
                 choices=idp_choices,
-                label=_("Identity Provider"),
+                label=_('Identity Provider'),
                 help_text=mark_safe(help_text),
             )
 
@@ -558,8 +496,8 @@ def clean(self):
         """
         super().clean()
 
-        provider_id = self.cleaned_data.get("provider_id", None)
-        enterprise_customer = self.cleaned_data.get("enterprise_customer", None)
+        provider_id = self.cleaned_data.get('provider_id', None)
+        enterprise_customer = self.cleaned_data.get('enterprise_customer', None)
 
         if provider_id is None or enterprise_customer is None:
             # field validation for either provider_id or enterprise_customer has already raised
@@ -594,23 +532,17 @@ def clean(self):
             )
 
         # There should be always one default provider in case an enterprise has multiple providers.
-        identity_provider_total_forms = int(
-            self.data.get("enterprise_customer_identity_providers-TOTAL_FORMS")
-        )
+        identity_provider_total_forms = int(self.data.get('enterprise_customer_identity_providers-TOTAL_FORMS'))
         default_provider_values = []
         for form_num in range(identity_provider_total_forms):
             default_provider_values.append(
-                self.data.get(
-                    "enterprise_customer_identity_providers-{}-default_provider".format(
-                        form_num
-                    )
-                )
+                self.data.get('enterprise_customer_identity_providers-{}-default_provider'.format(form_num))
             )
-        providers_marked_default = default_provider_values.count("on")
+        providers_marked_default = default_provider_values.count('on')
         if providers_marked_default > 1:
-            raise ValidationError("Please select only one default provider.")
+            raise ValidationError('Please select only one default provider.')
         if identity_provider_total_forms > 1 and providers_marked_default == 0:
-            raise ValidationError("Please select one default provider.")
+            raise ValidationError('Please select one default provider.')
 
 
 class EnterpriseCustomerReportingConfigAdminForm(forms.ModelForm):
@@ -620,7 +552,6 @@ class EnterpriseCustomerReportingConfigAdminForm(forms.ModelForm):
     This form uses the PasswordInput widget to obscure passwords as they are
     being entered by the user.
     """
-
     enterprise_customer_catalogs = forms.ModelMultipleChoiceField(
         EnterpriseCustomerCatalog.objects.all(),
         required=False,
@@ -651,8 +582,8 @@ class Meta:
             "enterprise_customer_catalogs",
         )
         widgets = {
-            "decrypted_password": forms.widgets.PasswordInput(),
-            "decrypted_sftp_password": forms.widgets.PasswordInput(),
+            'decrypted_password': forms.widgets.PasswordInput(),
+            'decrypted_sftp_password': forms.widgets.PasswordInput(),
         }
 
     def clean(self):
@@ -660,47 +591,45 @@ def clean(self):
         Override of clean method to perform additional validation
         """
         cleaned_data = super().clean()
-        report_customer = cleaned_data.get("enterprise_customer")
-        data_type = cleaned_data.get("data_type")
+        report_customer = cleaned_data.get('enterprise_customer')
+        data_type = cleaned_data.get('data_type')
 
-        if (
-            data_type in EnterpriseCustomerReportingConfiguration.PEARSON_ONLY_REPORTS
-            and report_customer.name != "Pearson"
-        ):
+        if data_type in EnterpriseCustomerReportingConfiguration.PEARSON_ONLY_REPORTS \
+                and report_customer.name != 'Pearson':
             message = _(
                 'This data_type "{data_type}" is not supported for enterprise'
-                "customer {enterprise_customer}. Please select a different data_type.",
+                'customer {enterprise_customer}. Please select a different data_type.',
             ).format(
                 enterprise_customer=report_customer,
                 data_type=data_type,
             )
-            self.add_error("data_type", message)
+            self.add_error('data_type', message)
 
         # Check that any selected catalogs are tied to the selected enterprise.
         invalid_catalogs = [
-            "{} ({})".format(catalog.title, catalog.uuid)
-            for catalog in cleaned_data.get("enterprise_customer_catalogs")
+            '{} ({})'.format(catalog.title, catalog.uuid)
+            for catalog in cleaned_data.get('enterprise_customer_catalogs')
             if catalog.enterprise_customer != report_customer
         ]
 
         if invalid_catalogs:
             message = _(
-                "These catalogs for reporting do not match enterprise"
-                "customer {enterprise_customer}: {invalid_catalogs}",
+                'These catalogs for reporting do not match enterprise'
+                'customer {enterprise_customer}: {invalid_catalogs}',
             ).format(
                 enterprise_customer=report_customer,
                 invalid_catalogs=invalid_catalogs,
             )
-            self.add_error("enterprise_customer_catalogs", message)
+            self.add_error('enterprise_customer_catalogs', message)
 
 
 class TransmitEnterpriseCoursesForm(forms.Form):
     """
     Form to transmit courses metadata for enterprise customers.
     """
-
     channel_worker_username = forms.CharField(
-        label=_("Enter enterprise channel worker username."), required=True
+        label=_('Enter enterprise channel worker username.'),
+        required=True
     )
 
     def clean_channel_worker_username(self):
@@ -710,7 +639,7 @@ def clean_channel_worker_username(self):
         Returns:
             str: the cleaned value of channel user username for transmitting courses metadata.
         """
-        channel_worker_username = self.cleaned_data["channel_worker_username"].strip()
+        channel_worker_username = self.cleaned_data['channel_worker_username'].strip()
 
         try:
             User.objects.get(username=channel_worker_username)
@@ -732,10 +661,10 @@ class SystemWideEnterpriseUserRoleAssignmentForm(UserRoleAssignmentAdminForm):
     class Meta:
         model = SystemWideEnterpriseUserRoleAssignment
         fields = [
-            "user",
-            "role",
-            "enterprise_customer",
-            "applies_to_all_contexts",
+            'user',
+            'role',
+            'enterprise_customer',
+            'applies_to_all_contexts',
         ]
 
     def __init__(self, *args, **kwargs):
@@ -746,15 +675,11 @@ def __init__(self, *args, **kwargs):
         """
         super().__init__(*args, **kwargs)
         try:
-            self.fields[
-                "enterprise_customer"
-            ].queryset = EnterpriseCustomer.objects.filter(
+            self.fields['enterprise_customer'].queryset = EnterpriseCustomer.objects.filter(
                 enterprise_customer_users__user_id=self.instance.user.id,
             )
         except SystemWideEnterpriseUserRoleAssignment.user.RelatedObjectDoesNotExist:
-            self.fields[
-                "enterprise_customer"
-            ].queryset = EnterpriseCustomer.objects.none()
+            self.fields['enterprise_customer'].queryset = EnterpriseCustomer.objects.none()
 
 
 class EnterpriseFeatureUserRoleAssignmentForm(UserRoleAssignmentAdminForm):
@@ -764,7 +689,7 @@ class EnterpriseFeatureUserRoleAssignmentForm(UserRoleAssignmentAdminForm):
 
     class Meta:
         model = EnterpriseFeatureUserRoleAssignment
-        fields = ["user", "role"]
+        fields = ['user', 'role']
 
 
 class AdminNotificationForm(forms.ModelForm):
@@ -774,9 +699,9 @@ class AdminNotificationForm(forms.ModelForm):
 
     class Meta:
         model = AdminNotification
-        fields = "__all__"
+        fields = '__all__'
         widgets = {
-            "text": forms.Textarea(attrs={"cols": 80, "rows": 5}),
+            'text': forms.Textarea(attrs={'cols': 80, 'rows': 5}),
         }
 
     def clean(self):
@@ -787,32 +712,28 @@ def clean(self):
         """
         cleaned_data = super().clean()
 
-        start_date = cleaned_data.get("start_date", None)
-        expiration_date = cleaned_data.get("expiration_date", None)
+        start_date = cleaned_data.get('start_date', None)
+        expiration_date = cleaned_data.get('expiration_date', None)
 
         if start_date is None or expiration_date is None:
             # Start and expiration dates are mandatory.
-            raise ValidationError("Start and expiration dates are mandatory.")
+            raise ValidationError('Start and expiration dates are mandatory.')
 
         if expiration_date < start_date:
             # `start_date` must always come before the `expiration_date`
-            raise ValidationError(
-                {"expiration_date": ["Expiration date should be after start date."]}
-            )
-        admin_notification = (
-            AdminNotification.objects.filter(
-                Q(start_date__range=(start_date, expiration_date))
-                | Q(expiration_date__range=(start_date, expiration_date))
-                | Q(start_date__lt=start_date, expiration_date__gt=expiration_date)
-            )
-            .exclude(pk=self.instance.id)
-            .exists()
-        )
+            raise ValidationError({'expiration_date': ['Expiration date should be after start date.']})
+        admin_notification = AdminNotification.objects.filter(
+            Q(start_date__range=(start_date, expiration_date)) |
+            Q(expiration_date__range=(start_date, expiration_date)) |
+            Q(start_date__lt=start_date, expiration_date__gt=expiration_date)
+        ).exclude(
+            pk=self.instance.id
+        ).exists()
 
         if admin_notification:
             # This should not happen, as there can be only one admin notification instance in a particular date range.
             message = _(
-                "Please select different date range. There is another notification scheduled in this date range.",
+                'Please select different date range. There is another notification scheduled in this date range.',
             )
             logger.exception(message)
 

From fe3017f5fe89d2f56b523ec351456d751e81ad42 Mon Sep 17 00:00:00 2001
From: Arunmozhi <arun@arunmozhi.in>
Date: Sun, 28 May 2023 16:29:37 +0530
Subject: [PATCH 4/9] feat: adds force_enrollment field to ManageLearnersForm

---
 enterprise/admin/forms.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/enterprise/admin/forms.py b/enterprise/admin/forms.py
index fddb2408ee..8d6c9234af 100644
--- a/enterprise/admin/forms.py
+++ b/enterprise/admin/forms.py
@@ -67,6 +67,11 @@ class ManageLearnersForm(forms.Form):
         label=_("Enroll these learners in this course"), required=False,
         help_text=_("To enroll learners in a course, enter a course ID."),
     )
+    force_enrollment = forms.BooleanField(
+        label=_("Force Enrollment"),
+        help_text=_("The selected course is 'Invite Only'. Only staff can enroll learners to this course."),
+        required=False,
+    )
     course_mode = forms.ChoiceField(
         label=_("Course enrollment track"), required=False,
         choices=BLANK_CHOICE_DASH + [
@@ -128,6 +133,7 @@ class Fields:
         REASON = "reason"
         SALES_FORCE_ID = "sales_force_id"
         DISCOUNT = "discount"
+        FORCE_ENROLLMENT = "force_enrollment"
 
     class CsvColumns:
         """

From 6fb8296a2ac7dc28338e7e52792960546368a767 Mon Sep 17 00:00:00 2001
From: Arunmozhi <arun@arunmozhi.in>
Date: Sun, 28 May 2023 16:34:30 +0530
Subject: [PATCH 5/9] chore: remove master branch gating for CI

---
 .github/workflows/ci.yml | 1 -
 1 file changed, 1 deletion(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 27428607a5..dbf7d46941 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -4,7 +4,6 @@ on:
   push:
     branches: [master]
   pull_request:
-    branches: [master]
 
 jobs:
   run_tests:

From d1438bcf59021c066be4f676cd78b47ab567ef80 Mon Sep 17 00:00:00 2001
From: Arunmozhi <arun@arunmozhi.in>
Date: Sun, 28 May 2023 16:46:14 +0530
Subject: [PATCH 6/9] fix: removed codecov from the ci requirements

---
 requirements/ci.in  |  1 -
 requirements/ci.txt | 22 ++++------------------
 2 files changed, 4 insertions(+), 19 deletions(-)

diff --git a/requirements/ci.in b/requirements/ci.in
index ec4b7ba2cd..d97da5dcbb 100644
--- a/requirements/ci.in
+++ b/requirements/ci.in
@@ -2,6 +2,5 @@
 
 -c constraints.txt
 
-codecov                   # Code coverage reporting
 tox                       # Virtualenv management for tests
 tox-battery==0.6.1        # Makes tox aware of requirements file changes
diff --git a/requirements/ci.txt b/requirements/ci.txt
index 84b7c6e70c..7754874e6f 100644
--- a/requirements/ci.txt
+++ b/requirements/ci.txt
@@ -1,25 +1,15 @@
 #
-# This file is autogenerated by pip-compile
+# This file is autogenerated by pip-compile with python 3.8
 # To update, run:
 #
-#    make upgrade
+#    pip-compile --no-emit-index-url --no-emit-trusted-host --output-file=requirements/ci.txt requirements/ci.in
 #
-certifi==2021.10.8
-    # via requests
-charset-normalizer==2.0.11
-    # via requests
-codecov==2.1.12
-    # via -r requirements/ci.in
-coverage==6.3.1
-    # via codecov
 distlib==0.3.4
     # via virtualenv
 filelock==3.4.2
     # via
     #   tox
     #   virtualenv
-idna==3.3
-    # via requests
 packaging==21.3
     # via tox
 platformdirs==2.5.0
@@ -30,21 +20,17 @@ py==1.11.0
     # via tox
 pyparsing==3.0.7
     # via packaging
-requests==2.27.1
-    # via codecov
 six==1.16.0
     # via
     #   tox
     #   virtualenv
 toml==0.10.2
     # via tox
-tox-battery==0.6.1
-    # via -r requirements/ci.in
 tox==3.24.5
     # via
     #   -r requirements/ci.in
     #   tox-battery
-urllib3==1.26.8
-    # via requests
+tox-battery==0.6.1
+    # via -r requirements/ci.in
 virtualenv==20.13.1
     # via tox

From bae18471b46cd10a9d9ce80b4023e2bfd39c50f8 Mon Sep 17 00:00:00 2001
From: Arunmozhi <arun@arunmozhi.in>
Date: Sun, 28 May 2023 16:53:53 +0530
Subject: [PATCH 7/9] Revert "fix: removed codecov from the ci requirements"

This reverts commit d1438bcf59021c066be4f676cd78b47ab567ef80.
---
 requirements/ci.in  |  1 +
 requirements/ci.txt | 22 ++++++++++++++++++----
 2 files changed, 19 insertions(+), 4 deletions(-)

diff --git a/requirements/ci.in b/requirements/ci.in
index d97da5dcbb..ec4b7ba2cd 100644
--- a/requirements/ci.in
+++ b/requirements/ci.in
@@ -2,5 +2,6 @@
 
 -c constraints.txt
 
+codecov                   # Code coverage reporting
 tox                       # Virtualenv management for tests
 tox-battery==0.6.1        # Makes tox aware of requirements file changes
diff --git a/requirements/ci.txt b/requirements/ci.txt
index 7754874e6f..84b7c6e70c 100644
--- a/requirements/ci.txt
+++ b/requirements/ci.txt
@@ -1,15 +1,25 @@
 #
-# This file is autogenerated by pip-compile with python 3.8
+# This file is autogenerated by pip-compile
 # To update, run:
 #
-#    pip-compile --no-emit-index-url --no-emit-trusted-host --output-file=requirements/ci.txt requirements/ci.in
+#    make upgrade
 #
+certifi==2021.10.8
+    # via requests
+charset-normalizer==2.0.11
+    # via requests
+codecov==2.1.12
+    # via -r requirements/ci.in
+coverage==6.3.1
+    # via codecov
 distlib==0.3.4
     # via virtualenv
 filelock==3.4.2
     # via
     #   tox
     #   virtualenv
+idna==3.3
+    # via requests
 packaging==21.3
     # via tox
 platformdirs==2.5.0
@@ -20,17 +30,21 @@ py==1.11.0
     # via tox
 pyparsing==3.0.7
     # via packaging
+requests==2.27.1
+    # via codecov
 six==1.16.0
     # via
     #   tox
     #   virtualenv
 toml==0.10.2
     # via tox
+tox-battery==0.6.1
+    # via -r requirements/ci.in
 tox==3.24.5
     # via
     #   -r requirements/ci.in
     #   tox-battery
-tox-battery==0.6.1
-    # via -r requirements/ci.in
+urllib3==1.26.8
+    # via requests
 virtualenv==20.13.1
     # via tox

From 9f8fb2ae9fbe4aac03b6b6cf6fa72e49046788f7 Mon Sep 17 00:00:00 2001
From: Arunmozhi <arun@arunmozhi.in>
Date: Sun, 28 May 2023 16:54:38 +0530
Subject: [PATCH 8/9] fix: set the codecov version to 2.1.13

---
 requirements/ci.txt | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/requirements/ci.txt b/requirements/ci.txt
index 84b7c6e70c..2b3e4e9a69 100644
--- a/requirements/ci.txt
+++ b/requirements/ci.txt
@@ -8,7 +8,7 @@ certifi==2021.10.8
     # via requests
 charset-normalizer==2.0.11
     # via requests
-codecov==2.1.12
+codecov==2.1.13
     # via -r requirements/ci.in
 coverage==6.3.1
     # via codecov

From 5c2db6b70ef558d24b5add77e19ccdca69136781 Mon Sep 17 00:00:00 2001
From: Arunmozhi <arun@arunmozhi.in>
Date: Tue, 30 May 2023 12:49:28 +0530
Subject: [PATCH 9/9] chore: bumped version and added changelog

---
 CHANGELOG.rst          | 4 ++++
 enterprise/__init__.py | 2 +-
 2 files changed, 5 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 9526237dda..e12f610d94 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -17,6 +17,10 @@ Unreleased
 ----------
 * nothing
 
+[3.42.7]
+--------
+feat: allow enrollment to invite-only courses via manage learners admin page
+
 [3.42.6]
 --------
 feat: allow enrollment api admin to see all enrollments
diff --git a/enterprise/__init__.py b/enterprise/__init__.py
index e063398869..decedbe1cb 100644
--- a/enterprise/__init__.py
+++ b/enterprise/__init__.py
@@ -2,6 +2,6 @@
 Your project description goes here.
 """
 
-__version__ = "3.42.6"
+__version__ = "3.42.7"
 
 default_app_config = "enterprise.apps.EnterpriseConfig"