Honor the enrollment mode in the Open edX enrollment webhook

openedx/views.py

@@ -16,6 +16,7 @@
 
 from courses.api import create_local_enrollment, generate_course_run_certificates
 from courses.models import CourseRun, CourseRunCertificate
+from openedx.constants import EDX_DEFAULT_ENROLLMENT_MODE
 from users.models import User
 
 log = logging.getLogger(__name__)
@@ -35,23 +36,29 @@ def edx_enrollment_webhook(request):
     """
     Webhook endpoint that receives enrollment notifications from Open edX.
 
-    When a user needs to be enrolled in a course (e.g., staff/instructor role added),
-    the Open edX plugin POSTs to this endpoint so MITx Online can enroll them as an
-    auditor in the corresponding course run.
+    When a user needs to be enrolled in a course (e.g., staff/instructor role added,
+    or a course team manually enrolls learners from the instructor dashboard), the
+    Open edX plugin POSTs to this endpoint so MITx Online can mirror the enrollment
+    in the corresponding course run.
 
     Authentication: OAuth2 Bearer token (Django OAuth Toolkit access token).
 
     Expected payload:
         {
             "email": "instructor@example.com",
             "course_id": "course-v1:MITx+1.001x+2025_T1",
-            "role": "instructor"
+            "role": "instructor",
+            "mode": "audit"
         }
+
+    Both "role" and "mode" are optional. "mode" defaults to the default edX
+    enrollment mode when it is absent or empty.
     """
     # --- Validate payload ---
     email = request.data.get("email")
     course_id = request.data.get("course_id")
     role = request.data.get("role", "")
+    mode = request.data.get("mode") or EDX_DEFAULT_ENROLLMENT_MODE
 
     if not email or not course_id:
         return Response(
@@ -91,7 +98,7 @@ def edx_enrollment_webhook(request):
 
     # --- Create local enrollment ---
     try:
-        enrollment, created = create_local_enrollment(user, course_run)
+        enrollment, created = create_local_enrollment(user, course_run, mode=mode)
     except Exception:
         log.exception(
             "Webhook: Error creating enrollment for user %s in course run %s",
@@ -104,9 +111,10 @@ def edx_enrollment_webhook(request):
         )
 
     log.info(
-        "Webhook: Successfully enrolled user %s in course run %s as auditor (role: %s, created: %s)",
+        "Webhook: Successfully enrolled user %s in course run %s (mode: %s, role: %s, created: %s)",
         email,
         course_id,
+        enrollment.enrollment_mode,
         role,
         created,
     )

openedx/views_test.py

@@ -242,6 +242,57 @@ def test_already_enrolled_user(self, api_client, oauth_token):
             == 1
         )
 
+    @pytest.mark.parametrize(
+        ("payload_mode", "expected_mode"),
+        [
+            (EDX_ENROLLMENT_VERIFIED_MODE, EDX_ENROLLMENT_VERIFIED_MODE),
+            (EDX_ENROLLMENT_AUDIT_MODE, EDX_ENROLLMENT_AUDIT_MODE),
+            (None, EDX_ENROLLMENT_AUDIT_MODE),
+            ("", EDX_ENROLLMENT_AUDIT_MODE),
+        ],
+    )
+    def test_enrollment_mode(
+        self, api_client, oauth_token, payload_mode, expected_mode
+    ):
+        """The payload mode is honored, falling back to the default mode"""
+        user = UserFactory.create()
+        course_run = CourseRunFactory.create()
+
+        payload = {
+            "email": user.email,
+            "course_id": course_run.courseware_id,
+        }
+        if payload_mode is not None:
+            payload["mode"] = payload_mode
+
+        response = self._post_webhook(api_client, payload, token=oauth_token.token)
+
+        assert response.status_code == status.HTTP_201_CREATED
+        enrollment = CourseRunEnrollment.all_objects.get(user=user, run=course_run)
+        assert enrollment.enrollment_mode == expected_mode
+
+    def test_mode_does_not_change_existing_enrollment(self, api_client, oauth_token):
+        """An existing enrollment keeps its mode; the webhook only mirrors state"""
+        user = UserFactory.create()
+        course_run = CourseRunFactory.create()
+        CourseRunEnrollment.all_objects.create(
+            user=user,
+            run=course_run,
+            edx_enrolled=True,
+            enrollment_mode=EDX_ENROLLMENT_VERIFIED_MODE,
+        )
+
+        payload = {
+            "email": user.email,
+            "course_id": course_run.courseware_id,
+            "mode": EDX_ENROLLMENT_AUDIT_MODE,
+        }
+        response = self._post_webhook(api_client, payload, token=oauth_token.token)
+
+        assert response.status_code == status.HTTP_200_OK
+        enrollment = CourseRunEnrollment.all_objects.get(user=user, run=course_run)
+        assert enrollment.enrollment_mode == EDX_ENROLLMENT_VERIFIED_MODE
+
     def test_no_edx_api_call(self, api_client, oauth_token):
         """Test that the webhook does NOT call back to edX API"""
         user = UserFactory.create()