Andal.LND/edx-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Jkantor/support (#21541)

Browse Source 
* refactor link_program_enrollments into it's own file, add support page
This commit is contained in:
Jansen Kantor
2019-09-05 13:35:43 -04:00
committed by GitHub
parent 6da1d061ff
commit 4988999697
11 changed files with 842 additions and 412 deletions
Download Patch File Download Diff File Expand all files Collapse all files

198
lms/djangoapps/program_enrollments/link_program_enrollments.py Normal file
View File

@@ -0,0 +1,198 @@
""" Function to link program enrollments and external_student_keys to an LMS user """
import logging
from uuid import UUID
from django.contrib.auth import get_user_model
from django.db import IntegrityError, transaction
from lms.djangoapps.program_enrollments.models import ProgramEnrollment
from student.models import CourseEnrollmentException
logger = logging.getLogger(__name__)
User = get_user_model()
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}'
)
@transaction.atomic
def link_program_enrollments_to_lms_users(program_uuid, external_keys_to_usernames):
u"""
Utility function to link ProgramEnrollments to LMS Users
Arguments:
-program_uuid: the program for which we are linking program enrollments
-external_keys_to_usernames: dict mapping `external_user_keys` to LMS usernames.
Returns:
{
(external_key, username): Error message if there was an error
}
Raises: ValueError if None is included in external_keys_to_usernames
This function will look up 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.
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 added to a dictionary of error messages keyed by
(external_key, username). The input will be skipped. All other inputs will be processed and
enrollments updated, and then the function will return the dictionary of error messages.
If there is an error while enrolling a user in a waiting program course enrollment, the
error will be logged, and added to the returned error dictionary, and we will roll back all
transactions for that user so that their db state will be the same as it was before this
function was called, to prevent program enrollments to be in a state where they have an LMS
user but still have waiting course enrollments. All other inputs will be processed
normally.
"""
_validate_inputs(program_uuid, external_keys_to_usernames)
errors = {}
program_enrollments = get_program_enrollments(program_uuid, external_keys_to_usernames.keys())
users = get_lms_users(external_keys_to_usernames.values())
for item in external_keys_to_usernames.items():
external_student_key, username = item
user = users.get(username)
error_message = None
if not user:
error_message = NO_LMS_USER_TPL.format(username)
program_enrollment = program_enrollments.get(external_student_key)
if not program_enrollment:
error_message = NO_PROGRAM_ENROLLMENT_TPL.format(
program_uuid=program_uuid,
external_student_key=external_student_key
)
elif program_enrollment.user:
error_message = get_existing_user_message(program_enrollment, user)
if error_message:
logger.warning(error_message)
errors[item] = error_message
continue
try:
with transaction.atomic():
link_program_enrollment_to_lms_user(program_enrollment, user)
except (CourseEnrollmentException, IntegrityError) as e:
logger.exception(u"Rolling back all operations for {}:{}".format(
external_student_key,
username,
))
error_message = type(e).__name__
if str(e):
error_message += ': '
error_message += str(e)
errors[item] = error_message
return errors
def _validate_inputs(program_uuid, external_keys_to_usernames):
if None in external_keys_to_usernames or None in external_keys_to_usernames.values():
raise ValueError('external_user_key or username cannot be None')
UUID(str(program_uuid)) # raises ValueError if invalid
def get_program_enrollments(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(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_to_lms_user(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
Raises: CourseEnrollmentException if there is an error enrolling user in a waiting
program course enrollment
IntegrityError if we try to create invalid records.
"""
try:
_link_program_enrollment(program_enrollment, user)
_link_course_enrollments(program_enrollment, user)
except IntegrityError:
logger.exception("Integrity error while linking program enrollments")
raise
def _link_program_enrollment(program_enrollment, user):
"""
Links program enrollment to user.
Raises IntegrityError if ProgramEnrollment is invalid
"""
logger.info(u'Linking external student key {} and user {}'.format(
program_enrollment.external_user_key,
user.username
))
program_enrollment.user = user
program_enrollment.save()
def _link_course_enrollments(program_enrollment, user):
"""
Enrolls user in waiting program course enrollments
Raises:
IntegrityError if a constraint is violated
CourseEnrollmentException if there is an issue enrolling the user in a course
"""
try:
for program_course_enrollment in program_enrollment.program_course_enrollments.all():
program_course_enrollment.enroll(user)
except CourseEnrollmentException as e:
error_message = COURSE_ENROLLMENT_ERR_TPL.format(
user=user.username,
course=program_course_enrollment.course_key
)
logger.exception(error_message)
raise type(e)(error_message)
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,
)

127
lms/djangoapps/program_enrollments/management/commands/link_program_enrollments.py
View File

@@ -3,21 +3,13 @@ import logging
from django.contrib.auth import get_user_model
from django.core.management.base import BaseCommand, CommandError
from django.db import IntegrityError, transaction
from lms.djangoapps.program_enrollments.models import ProgramEnrollment
from student.models import CourseEnrollmentException
from lms.djangoapps.program_enrollments.link_program_enrollments import link_program_enrollments_to_lms_users
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):
@@ -67,33 +59,12 @@ class Command(BaseCommand):
)
# 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
try:
with transaction.atomic():
self.link_program_enrollment(program_enrollment, user)
except (CourseEnrollmentException, IntegrityError):
logger.exception(u"Rolling back all operations for {}:{}".format(
external_student_key,
username,
))
continue # transaction rolled back
try:
link_program_enrollments_to_lms_users(program_uuid, ext_keys_to_usernames)
except Exception as e:
raise CommandError(e)
def parse_user_items(self, user_items):
"""
@@ -116,91 +87,3 @@ class Command(BaseCommand):
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
Raises: CourseEnrollmentException if there is an error enrolling user in a waiting
program course enrollment
IntegrityError if we try to create invalid records.
"""
try:
self._link_program_enrollment(program_enrollment, user)
self._link_course_enrollments(program_enrollment, user)
except IntegrityError:
logger.exception("Integrity error while linking program enrollments")
raise
def _link_program_enrollment(self, program_enrollment, user):
"""
Links program enrollment to user.
Raises IntegrityError if ProgramEnrollment is invalid
"""
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()
def _link_course_enrollments(self, program_enrollment, user):
"""
Enrolls user in waiting program course enrollments
Raises:
IntegrityError if a constraint is violated
CourseEnrollmentException if there is an issue enrolling the user in a course
"""
try:
for program_course_enrollment in program_enrollment.program_course_enrollments.all():
program_course_enrollment.enroll(user)
except CourseEnrollmentException:
logger.exception(COURSE_ENROLLMENT_ERR_TPL.format(
user=user.username,
course=program_course_enrollment.course_key
))
raise
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,
)

295
lms/djangoapps/program_enrollments/management/commands/tests/test_link_program_enrollments.py
View File

@@ -3,312 +3,27 @@ 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.tests.test_link_program_enrollments import TestLinkProgramEnrollmentsMixin
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.curriculum = 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,
curriculum_uuid=self.curriculum,
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 """
class TestLinkProgramEnrollmentManagementCommand(TestLinkProgramEnrollmentsMixin, TestCase):
""" Tests for exception behavior in the link_program_enrollments command """
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')
call_command(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')
course_enrollment_1 = self._create_waiting_course_enrollment(program_enrollment_1, nonexistant_course)
course_enrollment_2 = 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, self.fruit_course)
self._create_waiting_course_enrollment(program_enrollment_2, self.animal_course)
msg = COURSE_ENROLLMENT_ERR_TPL.format(user=self.user_1.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, 'ERROR', msg))
self._assert_no_program_enrollment(self.user_1, self.program)
self._assert_no_user(program_enrollment_1)
course_enrollment_1.refresh_from_db()
self.assertIsNone(course_enrollment_1.course_enrollment)
course_enrollment_2.refresh_from_db()
self.assertIsNone(course_enrollment_2.course_enrollment)
self._assert_user_enrolled_in_program_courses(self.user_2, self.program, self.animal_course, self.fruit_course)
def test_integrity_error(self):
existing_program_enrollment = self._create_waiting_enrollment(self.program, 'learner-0')
existing_program_enrollment.user = self.user_1
existing_program_enrollment.save()
program_enrollment_1 = self._create_waiting_enrollment(self.program, '0001')
self._create_waiting_enrollment(self.program, '0002')
msg = 'Integrity error while linking program enrollments'
with LogCapture() as logger:
self.call_command(self.program, ('0001', self.user_1.username), ('0002', self.user_2.username))
logger.check_present((COMMAND_PATH, 'ERROR', msg))
self._assert_no_user(program_enrollment_1)
self._assert_program_enrollment(self.user_2, self.program, '0002')
call_command(Command(), self.program, 'learner-01:user-01', 'learner-01:user-02')

370
lms/djangoapps/program_enrollments/tests/test_link_program_enrollments.py Normal file
View File

@@ -0,0 +1,370 @@
"""
Tests for link_program_enrollments.
"""
from __future__ import absolute_import
from uuid import uuid4
from testfixtures import LogCapture
from django.test import TestCase
from edx_django_utils.cache import RequestCache
from lms.djangoapps.program_enrollments.link_program_enrollments import (
link_program_enrollments_to_lms_users,
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
LOG_PATH = 'lms.djangoapps.program_enrollments.link_program_enrollments'
class TestLinkProgramEnrollmentsMixin(object):
""" Utility methods and test data for testing link_program_enrollments """
@classmethod
def setUpTestData(cls): # pylint: disable=missing-docstring
cls.program = uuid4()
cls.curriculum = 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 _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,
curriculum_uuid=self.curriculum,
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]
)
def _assert_error_message(self, errors, error_key, logger, log_level, expected_error_msg):
logger.check_present((LOG_PATH, log_level, expected_error_msg))
self.assertDictEqual(
{
error_key: expected_error_msg
},
errors
)
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)
link_program_enrollments_to_lms_users(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'
)
link_program_enrollments_to_lms_users(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_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:
errors = link_program_enrollments_to_lms_users(
self.program,
{
'0001': self.user_1.username,
'0002': self.user_2.username,
}
)
expected_error_msg = NO_PROGRAM_ENROLLMENT_TPL.format(
program_uuid=self.program,
external_student_key='0002'
)
logger.check_present((LOG_PATH, 'WARNING', expected_error_msg))
self.assertDictEqual(errors, {('0002', self.user_2.username): expected_error_msg})
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:
errors = link_program_enrollments_to_lms_users(
self.program,
{
'0001': self.user_1.username,
'0002': 'nonexistant-user',
}
)
expected_error_msg = NO_LMS_USER_TPL.format('nonexistant-user')
logger.check_present((LOG_PATH, 'WARNING', expected_error_msg))
self.assertDictEqual(errors, {('0002', 'nonexistant-user'): expected_error_msg})
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:
errors = link_program_enrollments_to_lms_users(
self.program,
{
'0001': self.user_1.username,
'0002': self.user_2.username
}
)
expected_error_msg = get_existing_user_message(program_enrollment, self.user_2)
logger.check_present((LOG_PATH, 'WARNING', expected_error_msg))
self.assertDictEqual(errors, {('0002', self.user_2.username): expected_error_msg})
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:
errors = link_program_enrollments_to_lms_users(
self.program,
{
'0001': self.user_1.username,
'0003': self.user_2.username,
}
)
expected_error_msg = get_existing_user_message(enrollment, self.user_2)
logger.check_present((LOG_PATH, 'WARNING', expected_error_msg))
self.assertDictEqual(errors, {('0003', self.user_2.username): expected_error_msg})
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')
course_enrollment_1 = self._create_waiting_course_enrollment(program_enrollment_1, nonexistant_course)
course_enrollment_2 = 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, self.fruit_course)
self._create_waiting_course_enrollment(program_enrollment_2, self.animal_course)
msg = COURSE_ENROLLMENT_ERR_TPL.format(user=self.user_1.username, course=nonexistant_course)
with LogCapture() as logger:
errors = link_program_enrollments_to_lms_users(
self.program,
{
'0001': self.user_1.username,
'0002': self.user_2.username
}
)
logger.check_present((LOG_PATH, 'ERROR', msg))
self.assertDictEqual(errors, {('0001', self.user_1.username): 'NonExistentCourseError: ' + msg})
self._assert_no_program_enrollment(self.user_1, self.program)
self._assert_no_user(program_enrollment_1)
course_enrollment_1.refresh_from_db()
self.assertIsNone(course_enrollment_1.course_enrollment)
course_enrollment_2.refresh_from_db()
self.assertIsNone(course_enrollment_2.course_enrollment)
self._assert_user_enrolled_in_program_courses(self.user_2, self.program, self.animal_course, self.fruit_course)
def test_integrity_error(self):
existing_program_enrollment = self._create_waiting_enrollment(self.program, 'learner-0')
existing_program_enrollment.user = self.user_1
existing_program_enrollment.save()
program_enrollment_1 = self._create_waiting_enrollment(self.program, '0001')
self._create_waiting_enrollment(self.program, '0002')
msg = 'Integrity error while linking program enrollments'
with LogCapture() as logger:
errors = link_program_enrollments_to_lms_users(
self.program,
{
'0001': self.user_1.username,
'0002': self.user_2.username,
}
)
logger.check_present((LOG_PATH, 'ERROR', msg))
self.assertEqual(len(errors), 1)
self.assertIn('UNIQUE constraint failed', errors[('0001', self.user_1.username)])
self._assert_no_user(program_enrollment_1)
self._assert_program_enrollment(self.user_2, self.program, '0002')
def test_invalid_uuid(self):
self._create_waiting_enrollment(self.program, 'learner-0')
with self.assertRaisesMessage(ValueError, 'badly formed hexadecimal UUID string'):
link_program_enrollments_to_lms_users(
'notauuid::thisisntauuid',
{
'learner-0': self.user_1.username,
}
)
def test_None(self):
self._create_waiting_enrollment(self.program, 'learner-0')
msg = 'external_user_key or username cannot be None'
with self.assertRaisesMessage(ValueError, msg):
link_program_enrollments_to_lms_users(
self.program,
{
None: self.user_1.username,
}
)
with self.assertRaisesMessage(ValueError, msg):
link_program_enrollments_to_lms_users(
'notauuid::thisisntauuid',
{
'learner-0': None,
}
)

55
lms/djangoapps/support/static/support/jsx/program_enrollments/index.jsx Normal file
View File

@@ -0,0 +1,55 @@
import React from 'react';
import PropTypes from 'prop-types';
import Cookies from 'js-cookie';
import { Button, InputText, TextArea, StatusAlert } from '@edx/paragon';
export const LinkProgramEnrollmentsSupportPage = props => (
<form method="post">
<input type="hidden" name="csrfmiddlewaretoken" value={Cookies.get('csrftoken')} />
{props.successes.length > 0 && (
<StatusAlert
open
alertType="success"
dialog={(
<div>
<span>There were { props.successes.length } successful linkages</span>
</div>
)}
/>
)}
{props.errors.map(errorItem => (
<StatusAlert
open
dismissible={false}
alertType="danger"
dialog={errorItem}
/>
))}
<InputText
name="program_uuid"
label="Program UUID"
value={props.programUUID}
/>
<TextArea
name="text"
label="List of external_user_key, lms_username, one per line"
value={props.text}
placeholder="external_student_key,lms_username"
/>
<Button label="Submit" type="submit" className={['btn', 'btn-primary']} />
</form>
);
LinkProgramEnrollmentsSupportPage.propTypes = {
successes: PropTypes.arrayOf(PropTypes.string),
errors: PropTypes.arrayOf(PropTypes.string),
programUUID: PropTypes.string,
text: PropTypes.string,
};
LinkProgramEnrollmentsSupportPage.defaultProps = {
successes: [],
errors: [],
programUUID: '',
text: '',
};

89
lms/djangoapps/support/tests/test_views.py
View File

@@ -9,6 +9,7 @@ import itertools
import json
import re
from datetime import datetime, timedelta
from uuid import uuid4
import ddt
import six
@@ -21,6 +22,7 @@ from pytz import UTC
from common.test.utils import disable_signal
from course_modes.models import CourseMode
from course_modes.tests.factories import CourseModeFactory
from lms.djangoapps.program_enrollments.link_program_enrollments import NO_PROGRAM_ENROLLMENT_TPL
from lms.djangoapps.verify_student.models import VerificationDeadline
from student.models import ENROLLED_TO_ENROLLED, CourseEnrollment, CourseEnrollmentAttribute, ManualEnrollmentAudit
from student.roles import GlobalStaff, SupportStaffRole
@@ -109,6 +111,7 @@ class SupportViewAccessTests(SupportViewTestCase):
'support:enrollment_list',
'support:manage_user',
'support:manage_user_detail',
'support:link_program_enrollments',
), (
(GlobalStaff, True),
(SupportStaffRole, True),
@@ -136,6 +139,7 @@ class SupportViewAccessTests(SupportViewTestCase):
"support:enrollment_list",
"support:manage_user",
"support:manage_user_detail",
"support:link_program_enrollments",
)
def test_require_login(self, url_name):
url = reverse(url_name)
@@ -160,6 +164,7 @@ class SupportViewIndexTests(SupportViewTestCase):
EXPECTED_URL_NAMES = [
"support:certificates",
"support:refund",
"support:link_program_enrollments",
]
def setUp(self):
@@ -434,3 +439,87 @@ class SupportViewEnrollmentsTests(SharedModuleStoreTestCase, SupportViewTestCase
)
verified_mode.expiration_datetime = datetime(year=1970, month=1, day=9, tzinfo=UTC)
verified_mode.save()
@ddt.ddt
class SupportViewLinkProgramEnrollmentsTests(SupportViewTestCase):
"""
Tests for the link_program_enrollments support view.
"""
def setUp(self):
"""Make the user support staff. """
super(SupportViewLinkProgramEnrollmentsTests, self).setUp()
self.url = reverse("support:link_program_enrollments")
SupportStaffRole().add_users(self.user)
self.program_uuid = str(uuid4())
self.text = '0001,user-0001\n0002,user-02'
def _assert_props(self, field_name, value, response):
self.assertIn('"{}": "{}"'.format(field_name, value), unicode(response.content, encoding='utf-8'))
def _assert_props_list(self, field_name, values, response):
"""
Assert that that page is being rendered with a specific list of props
"""
values_str = ''
if values:
values_str = '", "'.join(values)
values_str = '"{}"'.format(values_str)
self.assertIn(u'"{}": [{}]'.format(field_name, values_str), unicode(response.content, encoding='utf-8'))
def test_get(self):
response = self.client.get(self.url)
self._assert_props_list('successes', [], response)
self._assert_props_list('errors', [], response)
self._assert_props('programUUID', '', response)
self._assert_props('text', '', response)
def test_invalid_uuid(self):
response = self.client.post(self.url, data={
'program_uuid': 'notauuid',
'text': self.text,
})
self._assert_props_list('errors', [u'badly formed hexadecimal UUID string'], response)
@ddt.unpack
@ddt.data(
('program_uuid', ''),
('', 'text'),
('', ''),
)
def test_missing_parameter(self, program_uuid, text):
msg = u'You must provide both a program uuid and a comma separated list of external_student_key, username'
response = self.client.post(self.url, data={
'program_uuid': program_uuid,
'text': text,
})
self._assert_props_list('errors', [msg], response)
@ddt.data(
'0001,learner-01\n0002,learner-02', # normal
'0001,learner-01,apple,orange\n0002,learner-02,purple', # extra fields
'\t0001 , \t learner-01 \n 0002 , learner-02 ', # whitespace
)
@patch('support.views.program_enrollments.link_program_enrollments_to_lms_users')
def test_text(self, text, mocked_link):
self.client.post(self.url, data={
'program_uuid': self.program_uuid,
'text': text,
})
mocked_link.assert_called_once()
mocked_link.assert_called_with(
self.program_uuid,
{
'0001': 'learner-01',
'0002': 'learner-02',
}
)
def test_junk_text(self):
text = 'alsdjflajsdflakjs'
response = self.client.post(self.url, data={
'program_uuid': self.program_uuid,
'text': text,
})
msg = NO_PROGRAM_ENROLLMENT_TPL.format(program_uuid=self.program_uuid, external_student_key=text)
self._assert_props_list('errors', [msg], response)

2
lms/djangoapps/support/urls.py
View File

@@ -13,6 +13,7 @@ from support.views.feature_based_enrollments import FeatureBasedEnrollmentsSuppo
from support.views.index import index
from support.views.manage_user import ManageUserDetailView, ManageUserSupportView
from support.views.refund import RefundSupportView
from support.views.program_enrollments import LinkProgramEnrollmentSupportView
COURSE_ENTITLEMENTS_VIEW = EntitlementSupportView.as_view()
@@ -40,4 +41,5 @@ urlpatterns = [
FeatureBasedEnrollmentsSupportView.as_view(),
name="feature_based_enrollments"
),
url(r'link_program_enrollments/?$', LinkProgramEnrollmentSupportView.as_view(), name='link_program_enrollments')
]

5
lms/djangoapps/support/views/index.py
View File

@@ -43,6 +43,11 @@ SUPPORT_INDEX_URLS = [
"name": _("Feature Based Enrollments"),
"description": _("View feature based enrollment settings"),
},
{
"url": reverse_lazy("support:link_program_enrollments"),
"name": _("Link Program Enrollments"),
"description": _("Link LMS users to program enrollments"),
},
]

74
lms/djangoapps/support/views/program_enrollments.py Normal file
View File

@@ -0,0 +1,74 @@
"""
Support tool for changing course enrollments.
"""
from __future__ import absolute_import
import csv
from django.utils.decorators import method_decorator
from django.views.generic import View
from edxmako.shortcuts import render_to_response
from lms.djangoapps.support.decorators import require_support_permission
from lms.djangoapps.program_enrollments.link_program_enrollments import link_program_enrollments_to_lms_users
TEMPLATE_PATH = 'support/link_program_enrollments.html'
class LinkProgramEnrollmentSupportView(View):
"""
Allows viewing and changing learner enrollments by support
staff.
"""
# TODO: ARCH-91
# This view is excluded from Swagger doc generation because it
# does not specify a serializer class.
exclude_from_schema = True
@method_decorator(require_support_permission)
def get(self, request):
return render_to_response(
TEMPLATE_PATH,
{
'successes': [],
'errors': [],
'program_uuid': '',
'text': '',
}
)
@method_decorator(require_support_permission)
def post(self, request):
"""
Link the given program enrollments and lms users
"""
program_uuid = request.POST.get('program_uuid', '').strip()
text = request.POST.get('text', '')
successes = []
errors = []
if not program_uuid or not text:
error = 'You must provide both a program uuid and a comma separated list of external_student_key, username'
errors = [error]
else:
reader = csv.DictReader(text.splitlines(), fieldnames=('external_key', 'username'))
ext_key_to_lms_username = {
(item['external_key'] or '').strip(): (item['username'] or '').strip()
for item in reader
}
try:
link_errors = link_program_enrollments_to_lms_users(program_uuid, ext_key_to_lms_username)
except ValueError as e:
errors = [str(e)]
else:
successes = [str(item) for item in ext_key_to_lms_username.items() if item not in link_errors]
errors = [message for message in link_errors.values()]
return render_to_response(
TEMPLATE_PATH,
{
'successes': successes,
'errors': errors,
'program_uuid': program_uuid,
'text': text,
}
)

37
lms/templates/support/link_program_enrollments.html Normal file
View File

@@ -0,0 +1,37 @@
<%page expression_filter="h"/>
<%!
from django.utils.translation import ugettext as _
from openedx.core.djangolib.js_utils import js_escaped_string
%>
## Override the default styles_version to use Bootstrap
<%! main_css = "css/bootstrap/lms-main.css" %>
<%namespace name='static' file='../static_content.html'/>
<%inherit file="../main.html" />
<%block name="js_extra">
</%block>
<%block name="pagetitle">
${_("Link Program Enrollments")}
</%block>
<%block name="content">
<section class="container outside-app">
<h3> Link Program Enrollments </h3>
${static.renderReact(
component="LinkProgramEnrollmentsSupportPage",
id="entitlement-support-page",
props={
'successes': successes,
'errors': errors,
'text': text,
'programUUID': program_uuid
}
)
}
</section>
</%block>

2
webpack.common.config.js
View File

@@ -85,6 +85,8 @@ module.exports = Merge.smart({
SingleSupportForm: './lms/static/support/jsx/single_support_form.jsx',
AlertStatusBar: './lms/static/js/accessible_components/StatusBarAlert.jsx',
EntitlementSupportPage: './lms/djangoapps/support/static/support/jsx/entitlements/index.jsx',
LinkProgramEnrollmentsSupportPage: './lms/djangoapps/support/static/support/jsx/' +
'program_enrollments/index.jsx',
PasswordResetConfirmation: './lms/static/js/student_account/components/PasswordResetConfirmation.jsx',
StudentAccountDeletion: './lms/static/js/student_account/components/StudentAccountDeletion.jsx',
StudentAccountDeletionInitializer: './lms/static/js/student_account/StudentAccountDeletionInitializer.js',