Merge pull request #18583 from edx/LEARNER-5603-managment-command
LEARNER-5993 bulk change enrollment using csv file
This commit is contained in:
Ayub
@@ -0,0 +1,119 @@
|
||||
"""
|
||||
Management command to change many user enrollments in many courses using
|
||||
csv file.
|
||||
"""
|
||||
import csv
|
||||
import logging
|
||||
from os import path
|
||||
|
||||
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 student.models import CourseEnrollment, CourseEnrollmentAttribute, User
|
||||
|
||||
logger = logging.getLogger('student.management.commands.bulk_change_enrollment_csv')
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
"""
|
||||
Management command to change many user enrollments in many
|
||||
courses using the csv file
|
||||
"""
|
||||
|
||||
help = """
|
||||
Change the enrollment status of all the users specified in
|
||||
the csv file in the specified course to specified course
|
||||
mode.
|
||||
Could be used to update effected users by order
|
||||
placement issues. If large number of students are effected
|
||||
in different courses.
|
||||
Similar to bulk_change_enrollment but uses the csv file
|
||||
input format and can enroll students in multiple courses.
|
||||
|
||||
Example:
|
||||
$ ... bulk_change_enrollment_csv csv_file_path
|
||||
"""
|
||||
|
||||
def add_arguments(self, parser):
|
||||
""" Add argument to the command parser. """
|
||||
parser.add_argument(
|
||||
'--csv_file_path',
|
||||
required=True,
|
||||
help='Csv file path'
|
||||
)
|
||||
|
||||
def handle(self, *args, **options):
|
||||
""" Main handler for the command."""
|
||||
file_path = options['csv_file_path']
|
||||
|
||||
if not path.isfile(file_path):
|
||||
raise CommandError("File not found.")
|
||||
|
||||
with open(file_path) as csv_file:
|
||||
course_key = None
|
||||
user = None
|
||||
file_reader = csv.DictReader(csv_file)
|
||||
headers = file_reader.fieldnames
|
||||
|
||||
if not ('course_id' in headers and 'mode' in headers and 'user' in headers):
|
||||
raise CommandError('Invalid input CSV file.')
|
||||
|
||||
for row in file_reader:
|
||||
try:
|
||||
course_key = CourseKey.from_string(row['course_id'])
|
||||
except InvalidKeyError:
|
||||
logger.warning('Invalid or non-existent course id [{}]'.format(row['course_id']))
|
||||
|
||||
try:
|
||||
user = User.objects.get(username=row['user'])
|
||||
except:
|
||||
logger.warning('Invalid or non-existent user [{}]'.format(row['user']))
|
||||
if course_key and user:
|
||||
try:
|
||||
course_enrollment = CourseEnrollment.get_enrollment(user, course_key)
|
||||
# If student is not enrolled in course enroll the student in free mode
|
||||
if not course_enrollment:
|
||||
# try to create a enroll user in default course enrollment mode in case of
|
||||
# professional it will break because of no default course mode.
|
||||
try:
|
||||
course_enrollment = CourseEnrollment.get_or_create_enrollment(user=user,
|
||||
course_key=course_key)
|
||||
except Exception: # pylint: disable=broad-except
|
||||
# In case if no free mode is available.
|
||||
course_enrollment = None
|
||||
|
||||
if course_enrollment:
|
||||
# if student already had a enrollment and its mode is same as the provided one
|
||||
if course_enrollment.mode == row['mode']:
|
||||
logger.info("Student [%s] is already enrolled in Course [%s] in mode [%s].", user.username,
|
||||
course_key, course_enrollment.mode)
|
||||
# set the enrollment to active if its not already active.
|
||||
if not course_enrollment.is_active:
|
||||
course_enrollment.update_enrollment(is_active=True)
|
||||
else:
|
||||
# if student enrollment exists update it to new mode.
|
||||
with transaction.atomic():
|
||||
course_enrollment.update_enrollment(
|
||||
mode=row['mode'],
|
||||
is_active=True,
|
||||
skip_refund=True
|
||||
)
|
||||
course_enrollment.save()
|
||||
|
||||
if row['mode'] == 'credit':
|
||||
enrollment_attrs = [{
|
||||
'namespace': 'credit',
|
||||
'name': 'provider_id',
|
||||
'value': course_key.org,
|
||||
}]
|
||||
CourseEnrollmentAttribute.add_enrollment_attr(enrollment=course_enrollment,
|
||||
data_list=enrollment_attrs)
|
||||
else:
|
||||
# if student enrollment do not exists directly enroll in new mode.
|
||||
CourseEnrollment.enroll(user=user, course_key=course_key, mode=row['mode'])
|
||||
|
||||
except Exception as e:
|
||||
logger.info("Unable to update student [%s] course [%s] enrollment to mode [%s] "
|
||||
"because of Exception [%s]", row['user'], row['course_id'], row['mode'], repr(e))
|
||||
@@ -0,0 +1,117 @@
|
||||
from tempfile import NamedTemporaryFile
|
||||
import unittest
|
||||
|
||||
from django.conf import settings
|
||||
from django.core.management import call_command
|
||||
from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
|
||||
from xmodule.modulestore.tests.factories import CourseFactory
|
||||
from testfixtures import LogCapture
|
||||
|
||||
from course_modes.tests.factories import CourseModeFactory
|
||||
from course_modes.models import CourseMode
|
||||
from student.tests.factories import UserFactory
|
||||
from student.models import CourseEnrollment
|
||||
|
||||
|
||||
LOGGER_NAME = 'student.management.commands.bulk_change_enrollment_csv'
|
||||
|
||||
|
||||
class BulkChangeEnrollmentCSVTests(SharedModuleStoreTestCase):
|
||||
"""
|
||||
Tests bulk_change_enrollmetn_csv command
|
||||
"""
|
||||
def setUp(self):
|
||||
super(BulkChangeEnrollmentCSVTests, self).setUp()
|
||||
self.courses = []
|
||||
|
||||
self.user_info = [
|
||||
('amy', 'amy@pond.com', 'password'),
|
||||
('rory', 'rory@theroman.com', 'password'),
|
||||
('river', 'river@song.com', 'password')
|
||||
]
|
||||
|
||||
self.enrollments = []
|
||||
self.users = []
|
||||
|
||||
for username, email, password in self.user_info:
|
||||
user = UserFactory.create(username=username, email=email, password=password)
|
||||
self.users.append(user)
|
||||
course = CourseFactory.create()
|
||||
CourseModeFactory.create(course_id=course.id, mode_slug=CourseMode.AUDIT)
|
||||
CourseModeFactory.create(course_id=course.id, mode_slug=CourseMode.VERIFIED)
|
||||
self.courses.append(course)
|
||||
self.enrollments.append(CourseEnrollment.enroll(user, course.id, mode=CourseMode.AUDIT))
|
||||
|
||||
def _write_test_csv(self, csv, lines=None):
|
||||
"""Write a test csv file with the lines provided"""
|
||||
csv.write("course_id,user,mode,\n")
|
||||
csv.writelines(lines)
|
||||
csv.seek(0)
|
||||
return csv
|
||||
|
||||
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
|
||||
def test_user_not_exist(self):
|
||||
"""Verify that warning is logged for non existing user."""
|
||||
with NamedTemporaryFile() as csv:
|
||||
csv = self._write_test_csv(csv, lines="course-v1:edX+DemoX+Demo_Course,user,audit\n")
|
||||
|
||||
with LogCapture(LOGGER_NAME) as log:
|
||||
call_command("bulk_change_enrollment_csv", "--csv_file_path={}".format(csv.name))
|
||||
log.check(
|
||||
(
|
||||
LOGGER_NAME,
|
||||
'WARNING',
|
||||
'Invalid or non-existent user [user]'
|
||||
)
|
||||
)
|
||||
|
||||
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
|
||||
def test_invalid_course_key(self):
|
||||
"""Verify in case of invalid course key warning is logged."""
|
||||
with NamedTemporaryFile() as csv:
|
||||
csv = self._write_test_csv(csv, lines="Demo_Course,river,audit\n")
|
||||
|
||||
with LogCapture(LOGGER_NAME) as log:
|
||||
call_command("bulk_change_enrollment_csv", "--csv_file_path={}".format(csv.name))
|
||||
log.check(
|
||||
(
|
||||
LOGGER_NAME,
|
||||
'WARNING',
|
||||
'Invalid or non-existent course id [Demo_Course]'
|
||||
)
|
||||
)
|
||||
|
||||
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
|
||||
def test_already_enrolled_student(self):
|
||||
""" Verify in case if a user is already enrolled warning is logged."""
|
||||
with NamedTemporaryFile() as csv:
|
||||
csv = self._write_test_csv(csv, lines=str(self.courses[0].id) + ",amy,audit\n")
|
||||
|
||||
with LogCapture(LOGGER_NAME) as log:
|
||||
call_command("bulk_change_enrollment_csv", "--csv_file_path={}".format(csv.name))
|
||||
log.check(
|
||||
(
|
||||
LOGGER_NAME,
|
||||
'INFO',
|
||||
'Student [{}] is already enrolled in Course [{}] in mode [{}].'.format(
|
||||
'amy',
|
||||
str(self.courses[0].id),
|
||||
'audit',
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
|
||||
def test_bulk_enrollment(self):
|
||||
""" Test all users are enrolled using the command."""
|
||||
lines = (str(enrollment.course.id) + "," + str(enrollment.user.username) + ",verified\n"
|
||||
for enrollment in self.enrollments)
|
||||
|
||||
with NamedTemporaryFile() as csv:
|
||||
csv = self._write_test_csv(csv, lines=lines)
|
||||
call_command("bulk_change_enrollment_csv", "--csv_file_path={}".format(csv.name))
|
||||
|
||||
for enrollment in self.enrollments:
|
||||
new_enrollment = CourseEnrollment.get_enrollment(user=enrollment.user, course_key=enrollment.course)
|
||||
self.assertEqual(new_enrollment.is_active, True)
|
||||
self.assertEqual(new_enrollment.mode, CourseMode.VERIFIED)
|
||||
Reference in New Issue
Block a user