Upstream the Bulk Enroll API endpoint

lms/djangoapps/bulk_enroll/serializers.py

+"""
+Serializers for Bulk Enrollment.
+"""
+from opaque_keys import InvalidKeyError
+from opaque_keys.edx.keys import CourseKey
+from rest_framework import serializers
+
+
+class StringListField(serializers.ListField):
+    def to_internal_value(self, data):
+        try:
+            return data[0].split(',')
+        except IndexError:
+            return []
+
+
+class BulkEnrollmentSerializer(serializers.Serializer):
+    """Serializes enrollment information for a collection of students/emails.
+
+    This is mainly useful for implementing validation when performing bulk enrollment operations.
+    """
+    identifiers = serializers.CharField(required=True)
+    courses = StringListField(required=True)
+    action = serializers.ChoiceField(
+        choices=(
+            ('enroll', 'enroll'),
+            ('unenroll', 'unenroll')
+        ),
+        required=True
+    )
+    auto_enroll = serializers.BooleanField(default=False)
+    email_students = serializers.BooleanField(default=False)
+
+    def validate_courses(self, value):
+        """
+        Check that each course key in list is valid.
+        """
+        course_keys = value
+        for course in course_keys:
+            try:
+                CourseKey.from_string(course)
+            except InvalidKeyError:
+                raise serializers.ValidationError("Course key not valid: {}".format(course))
+        return value

lms/djangoapps/bulk_enroll/tests/test_views.py

+"""
+Tests for the Bulk Enrollment views.
+"""
+import json
+from django.conf import settings
+from django.contrib.auth.models import User
+from django.core import mail
+from django.core.urlresolvers import reverse
+from django.test.utils import override_settings
+from rest_framework.test import APIRequestFactory, APITestCase, force_authenticate
+
+from bulk_enroll.views import BulkEnrollView
+from courseware.tests.helpers import LoginEnrollmentTestCase
+from microsite_configuration import microsite
+from student.models import (
+    CourseEnrollment,
+    ManualEnrollmentAudit,
+    ENROLLED_TO_UNENROLLED,
+    UNENROLLED_TO_ENROLLED,
+)
+from student.tests.factories import UserFactory
+from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
+from xmodule.modulestore.tests.factories import CourseFactory
+
+
+@override_settings(ENABLE_BULK_ENROLLMENT_VIEW=True)
+class BulkEnrollmentTest(ModuleStoreTestCase, LoginEnrollmentTestCase, APITestCase):
+    """
+    Test the bulk enrollment endpoint
+    """
+
+    USERNAME = "Bob"
+    EMAIL = "bob@example.com"
+    PASSWORD = "edx"
+
+    def setUp(self):
+        """ Create a course and user, then log in. """
+        super(BulkEnrollmentTest, self).setUp()
+
+        self.view = BulkEnrollView.as_view()
+        self.request_factory = APIRequestFactory()
+        self.url = reverse('bulk_enroll')
+
+        self.staff = UserFactory.create(
+            username=self.USERNAME,
+            email=self.EMAIL,
+            password=self.PASSWORD,
+            is_staff=True,
+        )
+
+        self.course = CourseFactory.create()
+        self.course_key = unicode(self.course.id)
+        self.enrolled_student = UserFactory(username='EnrolledStudent', first_name='Enrolled', last_name='Student')
+        CourseEnrollment.enroll(
+            self.enrolled_student,
+            self.course.id
+        )
+        self.notenrolled_student = UserFactory(username='NotEnrolledStudent', first_name='NotEnrolled',
+                                               last_name='Student')
+
+        # Email URL values
+        self.site_name = microsite.get_value(
+            'SITE_NAME',
+            settings.SITE_NAME
+        )
+        self.about_path = '/courses/{}/about'.format(self.course.id)
+        self.course_path = '/courses/{}/'.format(self.course.id)
+
+    def request_bulk_enroll(self, data=None, **extra):
+        """ Make an authenticated request to the bulk enrollment API. """
+        request = self.request_factory.post(self.url, data=data, **extra)
+        force_authenticate(request, user=self.staff)
+        response = self.view(request)
+        response.render()
+        return response
+
+    def test_non_staff(self):
+        """ Test that non global staff users are forbidden from API use. """
+        self.staff.is_staff = False
+        self.staff.save()
+        response = self.request_bulk_enroll()
+        self.assertEqual(response.status_code, 403)
+
+    def test_missing_params(self):
+        """ Test the response when missing all query parameters. """
+        response = self.request_bulk_enroll()
+        self.assertEqual(response.status_code, 400)
+
+    def test_bad_action(self):
+        """ Test the response given an invalid action """
+        response = self.request_bulk_enroll({
+            'identifiers': self.enrolled_student.email,
+            'action': 'invalid-action',
+            'courses': self.course_key,
+        })
+        self.assertEqual(response.status_code, 400)
+
+    def test_invalid_email(self):
+        """ Test the response given an invalid email. """
+        response = self.request_bulk_enroll({
+            'identifiers': 'percivaloctavius@',
+            'action': 'enroll',
+            'email_students': False,
+            'courses': self.course_key,
+        })
+        self.assertEqual(response.status_code, 200)
+
+        # test the response data
+        expected = {
+            "action": "enroll",
+            'auto_enroll': False,
+            'email_students': False,
+            "courses": {
+                self.course_key: {
+                    "action": "enroll",
+                    'auto_enroll': False,
+                    "results": [
+                        {
+                            "identifier": 'percivaloctavius@',
+                            "invalidIdentifier": True,
+                        }
+                    ]
+                }
+            }
+        }
+
+        res_json = json.loads(response.content)
+        self.assertEqual(res_json, expected)
+
+    def test_invalid_username(self):
+        """ Test the response given an invalid username. """
+        response = self.request_bulk_enroll({
+            'identifiers': 'percivaloctavius',
+            'action': 'enroll',
+            'email_students': False,
+            'courses': self.course_key,
+        })
+        self.assertEqual(response.status_code, 200)
+
+        # test the response data
+        expected = {
+            "action": "enroll",
+            'auto_enroll': False,
+            'email_students': False,
+            "courses": {
+                self.course_key: {
+                    "action": "enroll",
+                    'auto_enroll': False,
+                    "results": [
+                        {
+                            "identifier": 'percivaloctavius',
+                            "invalidIdentifier": True,
+                        }
+                    ]
+                }
+            }
+        }
+
+        res_json = json.loads(response.content)
+        self.assertEqual(res_json, expected)
+
+    def test_enroll_with_username(self):
+        """ Test enrolling using a username as the identifier. """
+        response = self.request_bulk_enroll({
+            'identifiers': self.notenrolled_student.username,
+            'action': 'enroll',
+            'email_students': False,
+            'courses': self.course_key,
+        })
+        self.assertEqual(response.status_code, 200)
+
+        # test the response data
+        expected = {
+            "action": "enroll",
+            'auto_enroll': False,
+            "email_students": False,
+            "courses": {
+                self.course_key: {
+                    "action": "enroll",
+                    'auto_enroll': False,
+                    "results": [
+                        {
+                            "identifier": self.notenrolled_student.username,
+                            "before": {
+                                "enrollment": False,
+                                "auto_enroll": False,
+                                "user": True,
+                                "allowed": False,
+                            },
+                            "after": {
+                                "enrollment": True,
+                                "auto_enroll": False,
+                                "user": True,
+                                "allowed": False,
+                            }
+                        }
+                    ]
+                }
+            }
+        }
+        manual_enrollments = ManualEnrollmentAudit.objects.all()
+        self.assertEqual(manual_enrollments.count(), 1)
+        self.assertEqual(manual_enrollments[0].state_transition, UNENROLLED_TO_ENROLLED)
+        res_json = json.loads(response.content)
+        self.assertEqual(res_json, expected)
+
+    def test_enroll_with_email(self):
+        """ Test enrolling using a username as the identifier. """
+        response = self.request_bulk_enroll({
+            'identifiers': self.notenrolled_student.email,
+            'action': 'enroll',
+            'email_students': False,
+            'courses': self.course_key,
+        })
+        self.assertEqual(response.status_code, 200)
+
+        # test that the user is now enrolled
+        user = User.objects.get(email=self.notenrolled_student.email)
+        self.assertTrue(CourseEnrollment.is_enrolled(user, self.course.id))
+
+        # test the response data
+        expected = {
+            "action": "enroll",
+            "auto_enroll": False,
+            "email_students": False,
+            "courses": {
+                self.course_key: {
+                    "action": "enroll",
+                    "auto_enroll": False,
+                    "results": [
+                        {
+                            "identifier": self.notenrolled_student.email,
+                            "before": {
+                                "enrollment": False,
+                                "auto_enroll": False,
+                                "user": True,
+                                "allowed": False,
+                            },
+                            "after": {
+                                "enrollment": True,
+                                "auto_enroll": False,
+                                "user": True,
+                                "allowed": False,
+                            }
+                        }
+                    ]
+                }
+            }
+        }
+
+        manual_enrollments = ManualEnrollmentAudit.objects.all()
+        self.assertEqual(manual_enrollments.count(), 1)
+        self.assertEqual(manual_enrollments[0].state_transition, UNENROLLED_TO_ENROLLED)
+        res_json = json.loads(response.content)
+        self.assertEqual(res_json, expected)
+
+        # Check the outbox
+        self.assertEqual(len(mail.outbox), 0)
+
+    def test_unenroll(self):
+        """ Test unenrolling a user. """
+        response = self.request_bulk_enroll({'identifiers': self.enrolled_student.email, 'action': 'unenroll',
+                                             'email_students': False, 'courses': self.course_key, })
+        self.assertEqual(response.status_code, 200)
+
+        # test that the user is now unenrolled
+        user = User.objects.get(email=self.enrolled_student.email)
+        self.assertFalse(CourseEnrollment.is_enrolled(user, self.course.id))
+
+        # test the response data
+        expected = {
+            "action": "unenroll",
+            "auto_enroll": False,
+            "email_students": False,
+            "courses": {
+                self.course_key: {
+                    "action": "unenroll",
+                    "auto_enroll": False,
+                    "results": [
+                        {
+                            "identifier": self.enrolled_student.email,
+                            "before": {
+                                "enrollment": True,
+                                "auto_enroll": False,
+                                "user": True,
+                                "allowed": False,
+                            },
+                            "after": {
+                                "enrollment": False,
+                                "auto_enroll": False,
+                                "user": True,
+                                "allowed": False,
+                            }
+                        }
+                    ]
+                }
+            }
+
+        }
+
+        manual_enrollments = ManualEnrollmentAudit.objects.all()
+        self.assertEqual(manual_enrollments.count(), 1)
+        self.assertEqual(manual_enrollments[0].state_transition, ENROLLED_TO_UNENROLLED)
+        res_json = json.loads(response.content)
+        self.assertEqual(res_json, expected)
+
+        # Check the outbox
+        self.assertEqual(len(mail.outbox), 0)

lms/djangoapps/bulk_enroll/urls.py

+"""
+URLs for the Bulk Enrollment API
+"""
+from django.conf.urls import patterns, url
+
+from bulk_enroll.views import BulkEnrollView
+
+urlpatterns = patterns(
+    'bulk_enroll.views',
+    url(r'^bulk_enroll', BulkEnrollView.as_view(), name='bulk_enroll'),
+)

lms/djangoapps/bulk_enroll/views.py

+"""
+API views for Bulk Enrollment
+"""
+import json
+from edx_rest_framework_extensions.authentication import JwtAuthentication
+from rest_framework import status
+from rest_framework.response import Response
+from rest_framework.views import APIView
+
+from bulk_enroll.serializers import BulkEnrollmentSerializer
+from enrollment.views import EnrollmentUserThrottle
+from instructor.views.api import students_update_enrollment
+from openedx.core.lib.api.authentication import OAuth2Authentication
+from openedx.core.lib.api.permissions import IsStaff
+from util.disable_rate_limit import can_disable_rate_limit
+
+
+@can_disable_rate_limit
+class BulkEnrollView(APIView):
+    """
+    **Use Case**
+
+        Enroll multiple users in one or more courses.
+
+    **Example Request**
+
+        POST /api/bulk_enroll/v1/bulk_enroll/ {
+            "auto_enroll": true,
+            "email_students": true,
+            "action": "enroll",
+            "courses": "course-v1:edX+Demo+123,course-v1:edX+Demo2+456",
+            "identifiers": "brandon@example.com,yamilah@example.com"
+        }
+
+        **POST Parameters**
+
+          A POST request can include the following parameters.
+
+          * auto_enroll: When set to `true`, students will be enrolled as soon
+            as they register.
+          * email_students: When set to `true`, students will be sent email
+            notifications upon enrollment.
+          * action: Can either be set to "enroll" or "unenroll". This determines the behabior
+
+    **Response Values**
+
+        If the supplied course data is valid and the enrollments were
+        successful, an HTTP 200 "OK" response is returned.
+
+        The HTTP 200 response body contains a list of response data for each
+        enrollment. (See the `instructor.views.api.students_update_enrollment`
+        docstring for the specifics of the response data available for each
+        enrollment)
+    """
+
+    authentication_classes = JwtAuthentication, OAuth2Authentication
+    permission_classes = IsStaff,
+    throttle_classes = EnrollmentUserThrottle,
+
+    def post(self, request):
+        serializer = BulkEnrollmentSerializer(data=request.data)
+        if serializer.is_valid():
+            request.POST = request.data
+            response_dict = {
+                'auto_enroll': serializer.data.get('auto_enroll'),
+                'email_students': serializer.data.get('email_students'),
+                'action': serializer.data.get('action'),
+                'courses': {}
+            }
+            for course in serializer.data.get('courses'):
+                response = students_update_enrollment(self.request, course_id=course)
+                response_dict['courses'][course] = json.loads(response.content)
+            return Response(data=response_dict, status=status.HTTP_200_OK)
+        else:
+            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

lms/envs/common.py

@@ -392,6 +392,9 @@ FEATURES = {
     # Whether to check the "Notify users by email" checkbox in the batch enrollment form
     # in the instructor dashboard.
     'BATCH_ENROLLMENT_NOTIFY_USERS_DEFAULT': True,
+
+    # Whether the bulk enrollment view is enabled.
+    'ENABLE_BULK_ENROLLMENT_VIEW': False,
 }
 
 # Settings for the course reviews tool template and identification key, set either to None to disable course reviews
@@ -2110,6 +2113,9 @@ INSTALLED_APPS = (
     # Enrollment API
     'enrollment',
 
+    # Bulk Enrollment API
+    'bulk_enroll',
+
     # Student Identity Verification
     'lms.djangoapps.verify_student',
 

lms/envs/test.py

@@ -80,6 +80,8 @@ FEATURES['MILESTONES_APP'] = True
 
 FEATURES['ENABLE_ENROLLMENT_TRACK_USER_PARTITION'] = True
 
+FEATURES['ENABLE_BULK_ENROLLMENT_VIEW'] = True
+
 # Need wiki for courseware views to work. TODO (vshnayder): shouldn't need it.
 WIKI_ENABLED = True
 

lms/urls.py

@@ -806,6 +806,13 @@ if settings.FEATURES.get('RESTRICT_ENROLL_BY_REG_METHOD'):
 
     )
 
+if configuration_helpers.get_value('ENABLE_BULK_ENROLLMENT_VIEW',
+                                   settings.FEATURES['ENABLE_BULK_ENROLLMENT_VIEW']):
+    urlpatterns += (
+        url(r'^api/bulk_enroll/v1/', include('bulk_enroll.urls')),
+    )
+
+
 # Shopping cart
 urlpatterns += (
     url(r'^shoppingcart/', include('shoppingcart.urls')),

openedx/core/lib/api/permissions.py

@@ -119,6 +119,16 @@ class IsMasterCourseStaffInstructor(permissions.BasePermission):
         return False
 
 
+class IsStaff(permissions.BasePermission):
+    """
+    Permission that checks to see if the request user has is_staff access.
+    """
+
+    def has_permission(self, request, view):
+        if request.user.is_staff:
+            return True
+
+
 class IsUserInUrlOrStaff(IsUserInUrl):
     """
     Permission that checks to see if the request user matches the user in the URL or has is_staff access.