Add a bulk version of the change_enrollment script.

common/djangoapps/student/management/commands/bulk_change_enrollment.py

+"""Management command to change many user enrollments at once."""
+import logging
+
+from django.core.management.base import BaseCommand, CommandError
+from django.db import transaction
+from opaque_keys import InvalidKeyError
+from opaque_keys.edx.keys import CourseKey
+from optparse import make_option
+
+from course_modes.models import CourseMode
+from student.models import CourseEnrollment
+from xmodule.modulestore.django import modulestore
+
+logger = logging.getLogger(__name__)  # pylint: disable=invalid-name
+
+
+class Command(BaseCommand):
+    """Management command to change many user enrollments at once."""
+
+    help = """
+    Change the enrollment status for all users enrolled in a
+    particular mode for a course. Similar to the change_enrollment
+    script, but more useful for bulk moves.
+
+    Example:
+
+    Change enrollment for all audit users to honor in the given course.
+        $ ... bulk_change_enrollment -c course-v1:SomeCourse+SomethingX+2016 --from audit --to honor --commit
+
+    Without the --commit option, the command will have no effect.
+    """
+
+    option_list = BaseCommand.option_list + (
+        make_option(
+            '-f', '--from_mode',
+            dest='from',
+            default=None,
+            help='move from this enrollment mode'
+        ),
+        make_option(
+            '-t', '--to_mode',
+            dest='to',
+            default=None,
+            help='move to this enrollment mode'
+        ),
+        make_option(
+            '-c', '--course',
+            dest='course',
+            default=None,
+            help='the course to change enrollments in'
+        ),
+        make_option(
+            '--commit',
+            action='store_true',
+            dest='commit',
+            default=False,
+            help='display what will be done without any effect'
+        )
+    )
+
+    def handle(self, *args, **options):
+        course_id = options.get('course')
+        from_mode = options.get('from_mode')
+        to_mode = options.get('to_mode')
+        commit = options.get('commit')
+
+        if course_id is None:
+            raise CommandError('No course ID given.')
+        if from_mode is None or to_mode is None:
+            raise CommandError('Both `from` and `to` course modes must be given.')
+
+        try:
+            course_key = CourseKey.from_string(course_id)
+        except InvalidKeyError:
+            raise CommandError('Course ID {} is invalid.'.format(course_id))
+
+        if modulestore().get_course(course_key) is None:
+            raise CommandError('The given course {} does not exist.'.format(course_id))
+
+        if CourseMode.mode_for_course(course_key, to_mode) is None:
+            raise CommandError('The given mode to move users into ({}) does not exist.'.format(to_mode))
+
+        course_key_str = unicode(course_key)
+
+        try:
+            with transaction.atomic():
+                queryset = CourseEnrollment.objects.filter(course_id=course_key, mode=from_mode)
+                logger.info(
+                    'Moving %d users from %s to %s in course %s.', queryset.count(), from_mode, to_mode, course_key_str
+                )
+                queryset.update(mode=to_mode)
+
+                if not commit:
+                    raise Exception('The --commit flag was not given; forcing rollback.')
+                logger.info('Finished moving users from %s to %s in course %s.', from_mode, to_mode, course_key_str)
+        except Exception:  # pylint: disable=broad-except
+            logger.info('No users moved.')

common/djangoapps/student/management/tests/test_bulk_change_enrollment.py

+"""Tests for the bulk_change_enrollment command."""
+import ddt
+from django.core.management import call_command
+from django.core.management.base import CommandError
+
+from student.tests.factories import UserFactory, CourseModeFactory, CourseEnrollmentFactory
+from student.models import CourseEnrollment
+from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
+from xmodule.modulestore.tests.factories import CourseFactory
+
+
+@ddt.ddt
+class BulkChangeEnrollmentTests(SharedModuleStoreTestCase):
+    """Tests for the bulk_change_enrollment command."""
+
+    def setUp(self):
+        super(BulkChangeEnrollmentTests, self).setUp()
+        self.course = CourseFactory.create()
+        self.users = UserFactory.create_batch(5)
+
+    @ddt.data(('audit', 'honor'), ('honor', 'audit'))
+    @ddt.unpack
+    def test_bulk_convert(self, from_mode, to_mode):
+        """Verify that enrollments are changed correctly."""
+        self._enroll_users(from_mode)
+        CourseModeFactory(course_id=self.course.id, mode_slug=to_mode)
+
+        # Verify that no users are in the `from` mode yet.
+        self.assertEqual(len(CourseEnrollment.objects.filter(mode=to_mode, course_id=self.course.id)), 0)
+
+        call_command(
+            'bulk_change_enrollment',
+            course=unicode(self.course.id),
+            from_mode=from_mode,
+            to_mode=to_mode,
+            commit=True,
+        )
+
+        # Verify that all users have been moved -- if not, this will
+        # raise CourseEnrollment.DoesNotExist
+        for user in self.users:
+            CourseEnrollment.objects.get(mode=to_mode, course_id=self.course.id, user=user)
+
+    def test_without_commit(self):
+        """Verify that nothing happens when the `commit` flag is not given."""
+        self._enroll_users('audit')
+        CourseModeFactory(course_id=self.course.id, mode_slug='honor')
+
+        call_command(
+            'bulk_change_enrollment',
+            course=unicode(self.course.id),
+            from_mode='audit',
+            to_mode='honor',
+        )
+
+        # Verify that no users are in the honor mode.
+        self.assertEqual(len(CourseEnrollment.objects.filter(mode='honor', course_id=self.course.id)), 0)
+
+    def test_without_to_mode(self):
+        """Verify that the command fails when the `to_mode` argument does not exist."""
+        self._enroll_users('audit')
+        CourseModeFactory(course_id=self.course.id, mode_slug='audit')
+
+        with self.assertRaises(CommandError):
+            call_command(
+                'bulk_change_enrollment',
+                course=unicode(self.course.id),
+                from_mode='audit',
+                to_mode='honor',
+            )
+
+    @ddt.data('from_mode', 'to_mode', 'course')
+    def test_without_options(self, option):
+        """Verify that the command fails when some options are not given."""
+        command_options = {
+            'from_mode': 'audit',
+            'to_mode': 'honor',
+            'course': unicode(self.course.id),
+        }
+        command_options.pop(option)
+
+        with self.assertRaises(CommandError):
+            call_command('bulk_change_enrollment', **command_options)
+
+    def test_bad_course_id(self):
+        """Verify that the command fails when the given course ID does not parse."""
+        with self.assertRaises(CommandError):
+            call_command('bulk_change_enrollment', from_mode='audit', to_mode='honor', course='yolo', commit=True)
+
+    def test_nonexistent_course_id(self):
+        """Verify that the command fails when the given course does not exist."""
+        with self.assertRaises(CommandError):
+            call_command(
+                'bulk_change_enrollment',
+                from_mode='audit',
+                to_mode='honor',
+                course='course-v1:testX+test+2016',
+                commit=True
+            )
+
+    def _enroll_users(self, mode):
+        """Enroll users in the given mode."""
+        for user in self.users:
+            CourseEnrollmentFactory(mode=mode, course_id=self.course.id, user=user)