Jkantor/link program enrollments (#21330)
* add management command to link program enrollments to users
This commit is contained in:
Jansen Kantor
@@ -0,0 +1,170 @@
|
||||
""" Management command to link program enrollments and external student_keys to an LMS user """
|
||||
import logging
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.core.management.base import BaseCommand, CommandError
|
||||
from django.db import transaction
|
||||
from lms.djangoapps.program_enrollments.models import ProgramEnrollment
|
||||
from student.models import CourseEnrollmentException
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
User = get_user_model()
|
||||
|
||||
INCORRECT_PARAMETER_TPL = u'incorrectly formatted argument {}, must be in form <external user key>:<lms username>'
|
||||
DUPLICATE_KEY_TPL = u'external user key {} provided multiple times'
|
||||
NO_PROGRAM_ENROLLMENT_TPL = (u'No program enrollment found for program uuid={program_uuid} and external student '
|
||||
'key={external_student_key}')
|
||||
NO_LMS_USER_TPL = u'No user found with username {}'
|
||||
COURSE_ENROLLMENT_ERR_TPL = u'Failed to enroll user {user} with waiting program course enrollment for course {course}'
|
||||
EXISTING_USER_TPL = (u'Program enrollment with external_student_key={external_student_key} is already linked to '
|
||||
u'{account_relation} account username={username}')
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
"""
|
||||
Management command to manually link ProgramEnrollments without an LMS user to an LMS user by username
|
||||
|
||||
Usage:
|
||||
./manage.py lms link_program_enrollments <program_uuid> <user_item>*
|
||||
where a <user_item> is a string formatted as <external_user_key>:<lms_username>
|
||||
|
||||
Normally, program enrollments should be linked by the Django Social Auth post_save signal handler
|
||||
`lms.djangoapps.program_enrollments.signals.matriculate_learner`, but in the case that a partner does not
|
||||
have an IDP set up for learners to log in through, we need a way to link enrollments
|
||||
|
||||
Provided a program uuid and a list of external_user_key:lms_username, this command will look up the matching
|
||||
program enrollments and users, and update the program enrollments with the matching user. If the program
|
||||
enrollment has course enrollments, we will enroll the user into their waiting program courses.
|
||||
|
||||
If an external user key is specified twice, an exception will be raised and no enrollments will be modified.
|
||||
|
||||
For each external_user_key:lms_username, if:
|
||||
- The user is not found
|
||||
- No enrollment is found for the given program and external_user_key
|
||||
- The enrollment already has a user
|
||||
An error message will be logged and the input will be skipped. All other inputs will be processed and
|
||||
enrollments updated.
|
||||
|
||||
If there is an error while enrolling a user in a waiting program course enrollment, the error will be
|
||||
logged, but we will continue attempting to enroll the user in courses, and we will process all other
|
||||
input users
|
||||
"""
|
||||
|
||||
help = u'Manually links ProgramEnrollment records to LMS users'
|
||||
|
||||
def add_arguments(self, parser):
|
||||
parser.add_argument(
|
||||
'program_uuid',
|
||||
help='the program in which we are linking enrollments to users',
|
||||
)
|
||||
parser.add_argument(
|
||||
'user_items',
|
||||
nargs='*',
|
||||
help='specify the users to link, in the format <external_student_key>:<lms_username>*',
|
||||
)
|
||||
|
||||
# pylint: disable=arguments-differ
|
||||
@transaction.atomic
|
||||
def handle(self, program_uuid, user_items, *args, **options):
|
||||
ext_keys_to_usernames = self.parse_user_items(user_items)
|
||||
program_enrollments = self.get_program_enrollments(program_uuid, ext_keys_to_usernames.keys())
|
||||
users = self.get_lms_users(ext_keys_to_usernames.values())
|
||||
for external_student_key, username in ext_keys_to_usernames.items():
|
||||
program_enrollment = program_enrollments.get(external_student_key)
|
||||
if not program_enrollment:
|
||||
logger.warning(NO_PROGRAM_ENROLLMENT_TPL.format(
|
||||
program_uuid=program_uuid,
|
||||
external_student_key=external_student_key
|
||||
))
|
||||
continue
|
||||
|
||||
user = users.get(username)
|
||||
if not user:
|
||||
logger.warning(NO_LMS_USER_TPL.format(username))
|
||||
continue
|
||||
|
||||
self.link_program_enrollment(program_enrollment, user)
|
||||
|
||||
def parse_user_items(self, user_items):
|
||||
"""
|
||||
Params:
|
||||
list of strings in the format 'external_user_key:lms_username'
|
||||
Returns:
|
||||
dict mapping external user keys to lms usernames
|
||||
"""
|
||||
result = {}
|
||||
for user_item in user_items:
|
||||
split_args = user_item.split(':')
|
||||
if len(split_args) != 2:
|
||||
message = (INCORRECT_PARAMETER_TPL).format(user_item)
|
||||
raise CommandError(message)
|
||||
|
||||
external_user_key = split_args[0]
|
||||
lms_username = split_args[1]
|
||||
if external_user_key in result:
|
||||
raise CommandError(DUPLICATE_KEY_TPL.format(external_user_key))
|
||||
|
||||
result[external_user_key] = lms_username
|
||||
return result
|
||||
|
||||
def get_program_enrollments(self, program_uuid, external_student_keys):
|
||||
"""
|
||||
Does a bulk read of ProgramEnrollments for a given program and list of external student keys
|
||||
and returns a dict keyed by external student key
|
||||
"""
|
||||
program_enrollments = ProgramEnrollment.bulk_read_by_student_key(
|
||||
program_uuid,
|
||||
external_student_keys
|
||||
).prefetch_related(
|
||||
'program_course_enrollments'
|
||||
).select_related('user')
|
||||
return {
|
||||
program_enrollment.external_user_key: program_enrollment
|
||||
for program_enrollment in program_enrollments
|
||||
}
|
||||
|
||||
def get_lms_users(self, lms_usernames):
|
||||
"""
|
||||
Does a bulk read of Users by username and returns a dict keyed by username
|
||||
"""
|
||||
return {
|
||||
user.username: user
|
||||
for user in User.objects.filter(username__in=lms_usernames)
|
||||
}
|
||||
|
||||
def link_program_enrollment(self, program_enrollment, user):
|
||||
"""
|
||||
Attempts to link the given program enrollment to the given user
|
||||
If the enrollment has any program course enrollments, enroll the user in those courses as well
|
||||
"""
|
||||
if program_enrollment.user:
|
||||
logger.warning(get_existing_user_message(program_enrollment, user))
|
||||
return
|
||||
logger.info(u'Linking external student key {} and user {}'.format(
|
||||
program_enrollment.external_user_key,
|
||||
user.username
|
||||
))
|
||||
program_enrollment.user = user
|
||||
program_enrollment.save()
|
||||
|
||||
for program_course_enrollment in program_enrollment.program_course_enrollments.all():
|
||||
try:
|
||||
program_course_enrollment.enroll(user)
|
||||
except CourseEnrollmentException:
|
||||
logger.warning(COURSE_ENROLLMENT_ERR_TPL.format(
|
||||
user=user.username,
|
||||
course=program_course_enrollment.course_key
|
||||
))
|
||||
|
||||
|
||||
def get_existing_user_message(program_enrollment, user):
|
||||
"""
|
||||
Creates an error message that the specified program enrollment is already linked to an lms user
|
||||
"""
|
||||
existing_username = program_enrollment.user.username
|
||||
external_student_key = program_enrollment.external_user_key
|
||||
return EXISTING_USER_TPL.format(
|
||||
external_student_key=external_student_key,
|
||||
account_relation='target' if program_enrollment.user.id == user.id else 'a different',
|
||||
username=existing_username,
|
||||
)
|
||||
@@ -0,0 +1,292 @@
|
||||
"""
|
||||
Tests for the link_program_enrollments management command.
|
||||
"""
|
||||
from __future__ import absolute_import
|
||||
|
||||
from uuid import uuid4
|
||||
from testfixtures import LogCapture
|
||||
|
||||
from django.core.management import call_command
|
||||
from django.core.management.base import CommandError
|
||||
from django.test import TestCase
|
||||
|
||||
from edx_django_utils.cache import RequestCache
|
||||
from lms.djangoapps.program_enrollments.management.commands.link_program_enrollments import (
|
||||
Command,
|
||||
INCORRECT_PARAMETER_TPL,
|
||||
DUPLICATE_KEY_TPL,
|
||||
NO_PROGRAM_ENROLLMENT_TPL,
|
||||
NO_LMS_USER_TPL,
|
||||
COURSE_ENROLLMENT_ERR_TPL,
|
||||
get_existing_user_message,
|
||||
)
|
||||
from lms.djangoapps.program_enrollments.tests.factories import ProgramCourseEnrollmentFactory, ProgramEnrollmentFactory
|
||||
from opaque_keys.edx.keys import CourseKey
|
||||
from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
|
||||
from student.tests.factories import UserFactory
|
||||
|
||||
COMMAND_PATH = 'lms.djangoapps.program_enrollments.management.commands.link_program_enrollments'
|
||||
|
||||
|
||||
class TestLinkProgramEnrollmentsMixin(object):
|
||||
""" Utility methods and test data for testing the link_program_enrollments command """
|
||||
|
||||
@classmethod
|
||||
def setUpTestData(cls): # pylint: disable=missing-docstring
|
||||
cls.command = Command()
|
||||
cls.program = uuid4()
|
||||
cls.other_program = uuid4()
|
||||
cls.fruit_course = CourseKey.from_string('course-v1:edX+Oranges+Apples')
|
||||
cls.animal_course = CourseKey.from_string('course-v1:edX+Cats+Dogs')
|
||||
CourseOverviewFactory.create(id=cls.fruit_course)
|
||||
CourseOverviewFactory.create(id=cls.animal_course)
|
||||
|
||||
def setUp(self):
|
||||
self.user_1 = UserFactory.create()
|
||||
self.user_2 = UserFactory.create()
|
||||
|
||||
def tearDown(self):
|
||||
RequestCache.clear_all_namespaces()
|
||||
|
||||
def call_command(self, program_uuid, *user_info):
|
||||
"""
|
||||
Builds string arguments and calls the link_program_enrollments command
|
||||
"""
|
||||
command_args = [external_key + ":" + lms_username for external_key, lms_username in user_info]
|
||||
call_command(self.command, program_uuid, *command_args)
|
||||
|
||||
def _create_waiting_enrollment(self, program_uuid, external_user_key):
|
||||
"""
|
||||
Create a waiting program enrollment for the given program and external user key.
|
||||
"""
|
||||
return ProgramEnrollmentFactory.create(
|
||||
user=None,
|
||||
program_uuid=program_uuid,
|
||||
external_user_key=external_user_key,
|
||||
)
|
||||
|
||||
def _create_waiting_course_enrollment(self, program_enrollment, course_key, status='active'):
|
||||
"""
|
||||
Create a waiting program course enrollment for the given program enrollment, course key, and optionally status
|
||||
"""
|
||||
return ProgramCourseEnrollmentFactory.create(
|
||||
program_enrollment=program_enrollment,
|
||||
course_key=course_key,
|
||||
course_enrollment=None,
|
||||
status=status,
|
||||
)
|
||||
|
||||
def _assert_no_user(self, program_enrollment, refresh=True):
|
||||
"""
|
||||
Assert that the given program enrollment has no LMS user associated with it
|
||||
"""
|
||||
if refresh:
|
||||
program_enrollment.refresh_from_db()
|
||||
self.assertIsNone(program_enrollment.user)
|
||||
|
||||
def _assert_no_program_enrollment(self, user, program_uuid, refresh=True):
|
||||
"""
|
||||
Assert that the given user is not enrolled in the given program
|
||||
"""
|
||||
if refresh:
|
||||
user.refresh_from_db()
|
||||
self.assertFalse(user.programenrollment_set.filter(program_uuid=program_uuid).exists())
|
||||
|
||||
def _assert_program_enrollment(self, user, program_uuid, external_user_key, refresh=True):
|
||||
"""
|
||||
Assert that the given user is enrolled in the given program with the given external user key
|
||||
"""
|
||||
if refresh:
|
||||
user.refresh_from_db()
|
||||
enrollment = user.programenrollment_set.get(program_uuid=program_uuid, external_user_key=external_user_key)
|
||||
self.assertIsNotNone(enrollment)
|
||||
|
||||
def _assert_user_enrolled_in_program_courses(self, user, program_uuid, *course_keys):
|
||||
"""
|
||||
Assert that the given user is has active enrollments in the given courses through the given program
|
||||
"""
|
||||
user.refresh_from_db()
|
||||
program_enrollment = user.programenrollment_set.get(user=user, program_uuid=program_uuid)
|
||||
all_course_enrollments = program_enrollment.program_course_enrollments
|
||||
program_course_enrollments = all_course_enrollments.select_related(
|
||||
'course_enrollment__course'
|
||||
).filter(
|
||||
course_enrollment__isnull=False
|
||||
)
|
||||
course_enrollments = [
|
||||
program_course_enrollment.course_enrollment
|
||||
for program_course_enrollment in program_course_enrollments
|
||||
]
|
||||
self.assertTrue(all(course_enrollment.is_active for course_enrollment in course_enrollments))
|
||||
self.assertCountEqual(
|
||||
course_keys,
|
||||
[course_enrollment.course.id for course_enrollment in course_enrollments]
|
||||
)
|
||||
|
||||
|
||||
class TestLinkProgramEnrollments(TestLinkProgramEnrollmentsMixin, TestCase):
|
||||
""" Tests for link_program_enrollments behavior """
|
||||
|
||||
def test_link_only_specified_program(self):
|
||||
"""
|
||||
Test that when there are two waiting program enrollments with the same external user key,
|
||||
only the specified program's program enrollment will be linked
|
||||
"""
|
||||
program_enrollment = self._create_waiting_enrollment(self.program, '0001')
|
||||
self._create_waiting_course_enrollment(program_enrollment, self.fruit_course)
|
||||
self._create_waiting_course_enrollment(program_enrollment, self.animal_course)
|
||||
|
||||
another_program_enrollment = self._create_waiting_enrollment(self.other_program, '0001')
|
||||
self._create_waiting_course_enrollment(another_program_enrollment, self.fruit_course)
|
||||
self._create_waiting_course_enrollment(another_program_enrollment, self.animal_course)
|
||||
|
||||
self.call_command(self.program, ('0001', self.user_1.username))
|
||||
|
||||
self._assert_program_enrollment(self.user_1, self.program, '0001')
|
||||
self._assert_user_enrolled_in_program_courses(self.user_1, self.program, self.fruit_course, self.animal_course)
|
||||
|
||||
self._assert_no_user(another_program_enrollment)
|
||||
|
||||
def test_inactive_waiting_course_enrollment(self):
|
||||
"""
|
||||
Test that when a waiting program enrollment has waiting program course enrollments with a status of 'inactive'
|
||||
the course enrollment created after calling link_program_enrollments will be inactive
|
||||
"""
|
||||
program_enrollment = self._create_waiting_enrollment(self.program, '0001')
|
||||
active_enrollment = self._create_waiting_course_enrollment(
|
||||
program_enrollment,
|
||||
self.fruit_course
|
||||
)
|
||||
inactive_enrollment = self._create_waiting_course_enrollment(
|
||||
program_enrollment,
|
||||
self.animal_course,
|
||||
status='inactive'
|
||||
)
|
||||
|
||||
self.call_command(self.program, ('0001', self.user_1.username))
|
||||
|
||||
self._assert_program_enrollment(self.user_1, self.program, '0001')
|
||||
|
||||
active_enrollment.refresh_from_db()
|
||||
self.assertIsNotNone(active_enrollment.course_enrollment)
|
||||
self.assertEqual(active_enrollment.course_enrollment.course.id, self.fruit_course)
|
||||
self.assertTrue(active_enrollment.course_enrollment.is_active)
|
||||
|
||||
inactive_enrollment.refresh_from_db()
|
||||
self.assertIsNotNone(inactive_enrollment.course_enrollment)
|
||||
self.assertEqual(inactive_enrollment.course_enrollment.course.id, self.animal_course)
|
||||
self.assertFalse(inactive_enrollment.course_enrollment.is_active)
|
||||
|
||||
|
||||
class TestLinkProgramEnrollmentsErrors(TestLinkProgramEnrollmentsMixin, TestCase):
|
||||
""" Tests for link_program_enrollments error behavior """
|
||||
|
||||
def test_incorrectly_formatted_input(self):
|
||||
with self.assertRaisesRegex(CommandError, INCORRECT_PARAMETER_TPL.format('whoops')):
|
||||
call_command(self.command, self.program, 'learner-01:user-01', 'whoops', 'learner-03:user-03')
|
||||
|
||||
def test_repeated_user_key(self):
|
||||
with self.assertRaisesRegex(CommandError, DUPLICATE_KEY_TPL.format('learner-01')):
|
||||
self.call_command(self.program, ('learner-01', 'user-01'), ('learner-01', 'user-02'))
|
||||
|
||||
def test_program_enrollment_not_found__nonexistant(self):
|
||||
self._create_waiting_enrollment(self.program, '0001')
|
||||
self._program_enrollment_not_found()
|
||||
|
||||
def test_program_enrollment_not_found__different_program(self):
|
||||
self._create_waiting_enrollment(self.program, '0001')
|
||||
self._create_waiting_enrollment(self.other_program, '0002')
|
||||
self._program_enrollment_not_found()
|
||||
|
||||
def _program_enrollment_not_found(self):
|
||||
"""
|
||||
Helper for test_program_not_found_* tests.
|
||||
tries to link user_1 to '0001' and user_2 to '0002' in program
|
||||
asserts that user_2 was not linked because the enrollment was not found
|
||||
"""
|
||||
with LogCapture() as logger:
|
||||
self.call_command(self.program, ('0001', self.user_1.username), ('0002', self.user_2.username))
|
||||
logger.check_present(
|
||||
(COMMAND_PATH, 'WARNING', NO_PROGRAM_ENROLLMENT_TPL.format(
|
||||
program_uuid=self.program,
|
||||
external_student_key='0002'
|
||||
))
|
||||
)
|
||||
|
||||
self._assert_program_enrollment(self.user_1, self.program, '0001')
|
||||
self._assert_no_program_enrollment(self.user_2, self.program)
|
||||
|
||||
def test_user_not_found(self):
|
||||
self._create_waiting_enrollment(self.program, '0001')
|
||||
enrollment_2 = self._create_waiting_enrollment(self.program, '0002')
|
||||
|
||||
with LogCapture() as logger:
|
||||
self.call_command(self.program, ('0001', self.user_1.username), ('0002', 'nonexistant-user'))
|
||||
logger.check_present(
|
||||
(COMMAND_PATH, 'WARNING', NO_LMS_USER_TPL.format('nonexistant-user'))
|
||||
)
|
||||
|
||||
self._assert_program_enrollment(self.user_1, self.program, '0001')
|
||||
self._assert_no_user(enrollment_2)
|
||||
|
||||
def test_enrollment_already_linked_to_target_user(self):
|
||||
self._create_waiting_enrollment(self.program, '0001')
|
||||
program_enrollment = ProgramEnrollmentFactory.create(
|
||||
user=self.user_2,
|
||||
program_uuid=self.program,
|
||||
external_user_key='0002',
|
||||
)
|
||||
self._assert_no_program_enrollment(self.user_1, self.program, refresh=False)
|
||||
self._assert_program_enrollment(self.user_2, self.program, '0002', refresh=False)
|
||||
|
||||
with LogCapture() as logger:
|
||||
self.call_command(self.program, ('0001', self.user_1.username), ('0002', self.user_2.username))
|
||||
logger.check_present(
|
||||
(COMMAND_PATH, 'WARNING', get_existing_user_message(program_enrollment, self.user_2))
|
||||
)
|
||||
|
||||
self._assert_program_enrollment(self.user_1, self.program, '0001')
|
||||
self._assert_program_enrollment(self.user_2, self.program, '0002')
|
||||
|
||||
def test_enrollment_already_linked_to_different_user(self):
|
||||
self._create_waiting_enrollment(self.program, '0001')
|
||||
enrollment = ProgramEnrollmentFactory.create(
|
||||
program_uuid=self.program,
|
||||
external_user_key='0003',
|
||||
)
|
||||
user_3 = enrollment.user
|
||||
|
||||
self._assert_no_program_enrollment(self.user_1, self.program, refresh=False)
|
||||
self._assert_no_program_enrollment(self.user_2, self.program, refresh=False)
|
||||
self._assert_program_enrollment(user_3, self.program, '0003', refresh=False)
|
||||
|
||||
with LogCapture() as logger:
|
||||
self.call_command(self.program, ('0001', self.user_1.username), ('0003', self.user_2.username))
|
||||
logger.check_present(
|
||||
(COMMAND_PATH, 'WARNING', get_existing_user_message(enrollment, self.user_2))
|
||||
)
|
||||
|
||||
self._assert_program_enrollment(self.user_1, self.program, '0001')
|
||||
self._assert_no_program_enrollment(self.user_2, self.program)
|
||||
self._assert_program_enrollment(user_3, self.program, '0003')
|
||||
|
||||
def test_error_enrolling_in_course(self):
|
||||
nonexistant_course = CourseKey.from_string('course-v1:edX+Zilch+Bupkis')
|
||||
|
||||
program_enrollment_1 = self._create_waiting_enrollment(self.program, '0001')
|
||||
self._create_waiting_course_enrollment(program_enrollment_1, nonexistant_course)
|
||||
self._create_waiting_course_enrollment(program_enrollment_1, self.animal_course)
|
||||
|
||||
program_enrollment_2 = self._create_waiting_enrollment(self.program, '0002')
|
||||
self._create_waiting_course_enrollment(program_enrollment_2, nonexistant_course)
|
||||
self._create_waiting_course_enrollment(program_enrollment_2, self.animal_course)
|
||||
|
||||
msg_1 = COURSE_ENROLLMENT_ERR_TPL.format(user=self.user_1.username, course=nonexistant_course)
|
||||
msg_2 = COURSE_ENROLLMENT_ERR_TPL.format(user=self.user_2.username, course=nonexistant_course)
|
||||
with LogCapture() as logger:
|
||||
self.call_command(self.program, ('0001', self.user_1.username), ('0002', self.user_2.username))
|
||||
logger.check_present((COMMAND_PATH, 'WARNING', msg_1))
|
||||
logger.check_present((COMMAND_PATH, 'WARNING', msg_2))
|
||||
|
||||
self._assert_user_enrolled_in_program_courses(self.user_1, self.program, self.animal_course)
|
||||
self._assert_user_enrolled_in_program_courses(self.user_2, self.program, self.animal_course)
|
||||
Reference in New Issue
Block a user