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

Create Python API for program_enrollments: Part III

Browse Source 
This is the third in a series of commits to create
a Python API for the LMS program_enrollments app.
It does the following:
* Creates api/ folder.
* Moves link_program_enrollments.py to api/linking.py
* Creates api/reading.py for enrollment-fetching
  functions.
* Updates rest of app to use api/reading.py when
  it was going directly through the models before.
* Other misc. cleanup (isorting, unicode_literals,
  line breaks, etc).

Still to do:
* Create api/writing.py and update app to use it instead
  of going directly through models.
* Create api/reset.py and api/expire.py, which the management
  commands call out to.

EDUCATOR-4321
This commit is contained in:
Kyle McCormick
2019-09-05 12:02:01 -04:00
committed by Kyle McCormick
parent 77aacee6ed
commit 358f989131
26 changed files with 1058 additions and 279 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
lms/djangoapps/program_enrollments/api/__init__.py
View File

@@ -0,0 +1,11 @@
"""
Python API exposed by the proram_enrollments app to other in-process apps.
The functions are split into separate files for code organization, but they
are wildcard-imported into here so they can be imported directly from
`lms.djangoapps.program_enrollments.api`.
"""
from __future__ import absolute_import
from .linking import * # pylint: disable=wildcard-import
from .reading import * # pylint: disable=wildcard-import

10
lms/djangoapps/program_enrollments/api/api.py
View File

@@ -2,17 +2,17 @@
"""
ProgramEnrollment internal API intended for Enterprise API.
This is not part of the program_enrollments Python API.
The Enterprise API currently depends on this module being present with these
functions, as implemented in ./utils.py. This module will be refactored
away in https://openedx.atlassian.net/browse/ENT-2294
"""
from __future__ import absolute_import, unicode_literals
from lms.djangoapps.program_enrollments.rest_api.v1.utils import (
get_due_dates as get_due_dates_util,
get_course_run_url as get_course_run_url_util,
get_emails_enabled as get_emails_enabled_util,
)
from lms.djangoapps.program_enrollments.rest_api.v1.utils import get_course_run_url as get_course_run_url_util
from lms.djangoapps.program_enrollments.rest_api.v1.utils import get_due_dates as get_due_dates_util
from lms.djangoapps.program_enrollments.rest_api.v1.utils import get_emails_enabled as get_emails_enabled_util
def get_due_dates(request, course_key, user):

211
lms/djangoapps/program_enrollments/api/linking.py Normal file
View File

@@ -0,0 +1,211 @@
"""
Python API function to link program enrollments and external_student_keys to an
LMS user.
Outside of this subpackage, import these functions
from `lms.djangoapps.program_enrollments.api`.
"""
from __future__ import absolute_import, unicode_literals
import logging
from uuid import UUID
from django.contrib.auth import get_user_model
from django.db import IntegrityError, transaction
from student.models import CourseEnrollmentException
from .reading import fetch_program_enrollments
logger = logging.getLogger(__name__)
User = get_user_model()
NO_PROGRAM_ENROLLMENT_TEMPLATE = (
'No program enrollment found for program uuid={program_uuid} and external student '
'key={external_student_key}'
)
NO_LMS_USER_TEMPLATE = 'No user found with username {}'
COURSE_ENROLLMENT_ERR_TEMPLATE = (
'Failed to enroll user {user} with waiting program course enrollment for course {course}'
)
EXISTING_USER_TEMPLATE = (
'Program enrollment with external_student_key={external_student_key} is already linked to '
'{account_relation} account username={username}'
)
@transaction.atomic
def link_program_enrollments_to_lms_users(program_uuid, external_keys_to_usernames):
"""
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_by_ext_key(
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_TEMPLATE.format(username)
program_enrollment = program_enrollments.get(external_student_key)
if not program_enrollment:
error_message = NO_PROGRAM_ENROLLMENT_TEMPLATE.format(
program_uuid=program_uuid,
external_student_key=external_student_key
)
elif program_enrollment.user:
error_message = user_already_linked_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("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 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 user_already_linked_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_TEMPLATE.format(
external_student_key=external_student_key,
account_relation='target' if program_enrollment.user.id == user.id else 'a different',
username=existing_username,
)
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_by_ext_key(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 = fetch_program_enrollments(
program_uuid=program_uuid,
external_user_keys=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(program_enrollment, user):
"""
Links program enrollment to user.
Raises IntegrityError if ProgramEnrollment is invalid
"""
logger.info('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_TEMPLATE.format(
user=user.username,
course=program_course_enrollment.course_key
)
logger.exception(error_message)
raise type(e)(error_message)

321
lms/djangoapps/program_enrollments/api/reading.py Normal file
View File

@@ -0,0 +1,321 @@
"""
Python API functions related to reading program enrollments.
Outside of this subpackage, import these functions
from `lms.djangoapps.program_enrollments.api`.
"""
from __future__ import absolute_import, unicode_literals
from ..models import ProgramCourseEnrollment, ProgramEnrollment
_STUDENT_ARG_ERROR_MESSAGE = (
"user and external_user_key are both None; at least one must be provided."
)
_REALIZED_FILTER_ERROR_TEMPLATE = (
"{} and {} are mutually exclusive; at most one of them may be passed in as True."
)
def get_program_enrollment(
program_uuid,
user=None,
external_user_key=None,
curriculum_uuid=None,
):
"""
Get a single program enrollment.
Required arguments:
* program_uuid (UUID|str)
* At least one of:
* user (User)
* external_user_key (str)
Optional arguments:
* curriculum_uuid (UUID|str) [optional]
Returns: ProgramEnrollment
Raises: ProgramEnrollment.DoesNotExist, ProgramEnrollment.MultipleObjectsReturned
"""
if not (user or external_user_key):
raise ValueError(_STUDENT_ARG_ERROR_MESSAGE)
filters = {
"user": user,
"external_user_key": external_user_key,
"curriculum_uuid": curriculum_uuid,
}
return ProgramEnrollment.objects.get(
program_uuid=program_uuid, **_remove_none_values(filters)
)
def get_program_course_enrollment(
program_uuid,
course_key,
user=None,
external_user_key=None,
curriculum_uuid=None,
):
"""
Get a single program-course enrollment.
Required arguments:
* program_uuid (UUID|str)
* course_key (CourseKey|str)
* At least one of:
* user (User)
* external_user_key (str)
Optional arguments:
* curriculum_uuid (UUID|str) [optional]
Returns: ProgramCourseEnrollment
Raises:
* ProgramCourseEnrollment.DoesNotExist
* ProgramCourseEnrollment.MultipleObjectsReturned
"""
if not (user or external_user_key):
raise ValueError(_STUDENT_ARG_ERROR_MESSAGE)
filters = {
"program_enrollment__user": user,
"program_enrollment__external_user_key": external_user_key,
"program_enrollment__curriculum_uuid": curriculum_uuid,
}
return ProgramCourseEnrollment.objects.get(
program_enrollment__program_uuid=program_uuid,
course_key=course_key,
**_remove_none_values(filters)
)
def fetch_program_enrollments(
program_uuid,
curriculum_uuids=None,
users=None,
external_user_keys=None,
program_enrollment_statuses=None,
realized_only=False,
waiting_only=False,
):
"""
Fetch program enrollments for a specific program.
Required argument:
* program_uuid (UUID|str)
Optional arguments:
* curriculum_uuids (iterable[UUID|str])
* users (iterable[User])
* external_user_keys (iterable[str])
* program_enrollment_statuses (iterable[str])
* realized_only (bool)
* waiting_only (bool)
Optional arguments are used as filtersets if they are not None.
At most one of (realized_only, waiting_only) may be provided.
Returns: queryset[ProgramEnrollment]
"""
if realized_only and waiting_only:
raise ValueError(
_REALIZED_FILTER_ERROR_TEMPLATE.format("realized_only", "waiting_only")
)
filters = {
"curriculum_uuid__in": curriculum_uuids,
"user__in": users,
"external_user_key__in": external_user_keys,
"status__in": program_enrollment_statuses,
}
if realized_only:
filters["user__isnull"] = False
if waiting_only:
filters["user__isnull"] = True
return ProgramEnrollment.objects.filter(
program_uuid=program_uuid, **_remove_none_values(filters)
)
def fetch_program_course_enrollments(
program_uuid,
course_key,
curriculum_uuids=None,
users=None,
external_user_keys=None,
program_enrollment_statuses=None,
active_only=False,
inactive_only=False,
realized_only=False,
waiting_only=False,
):
"""
Fetch program-course enrollments for a specific program and course run.
Required argument:
* program_uuid (UUID|str)
* course_key (CourseKey|str)
Optional arguments:
* curriculum_uuids (iterable[UUID|str])
* users (iterable[User])
* external_user_keys (iterable[str])
* program_enrollment_statuses (iterable[str])
* active_only (bool)
* inactive_only (bool)
* realized_only (bool)
* waiting_only (bool)
Optional arguments are used as filtersets if they are not None.
At most one of (realized_only, waiting_only) may be provided.
At most one of (active_only, inactive_only) may be provided.
Returns: queryset[ProgramCourseEnrollment]
"""
if active_only and inactive_only:
raise ValueError(
_REALIZED_FILTER_ERROR_TEMPLATE.format("active_only", "inactive_only")
)
if realized_only and waiting_only:
raise ValueError(
_REALIZED_FILTER_ERROR_TEMPLATE.format("realized_only", "waiting_only")
)
filters = {
"program_enrollment__curriculum_uuid__in": curriculum_uuids,
"program_enrollment__user__in": users,
"program_enrollment__external_user_key__in": external_user_keys,
"program_enrollment__status__in": program_enrollment_statuses,
}
if active_only:
filters["status"] = "active"
if inactive_only:
filters["status"] = "inactive"
if realized_only:
filters["program_enrollment__user__isnull"] = False
if waiting_only:
filters["program_enrollment__user__isnull"] = True
return ProgramCourseEnrollment.objects.filter(
program_enrollment__program_uuid=program_uuid,
course_key=course_key,
**_remove_none_values(filters)
)
def fetch_program_enrollments_by_student(
user=None,
external_user_key=None,
program_uuids=None,
curriculum_uuids=None,
program_enrollment_statuses=None,
realized_only=False,
waiting_only=False,
):
"""
Fetch program enrollments for a specific student.
Required arguments (at least one must be provided):
* user (User)
* external_user_key (str)
Optional arguments:
* provided_uuids (iterable[UUID|str])
* curriculum_uuids (iterable[UUID|str])
* program_enrollment_statuses (iterable[str])
* realized_only (bool)
* waiting_only (bool)
Optional arguments are used as filtersets if they are not None.
At most one of (realized_only, waiting_only) may be provided.
Returns: queryset[ProgramEnrollment]
"""
if not (user or external_user_key):
raise ValueError(_STUDENT_ARG_ERROR_MESSAGE)
if realized_only and waiting_only:
raise ValueError(
_REALIZED_FILTER_ERROR_TEMPLATE.format("realized_only", "waiting_only")
)
filters = {
"user": user,
"external_user_key": external_user_key,
"program_uuid__in": program_uuids,
"curriculum_uuid__in": curriculum_uuids,
"status__in": program_enrollment_statuses,
}
if realized_only:
filters["user__isnull"] = False
if waiting_only:
filters["user__isnull"] = True
return ProgramEnrollment.objects.filter(**_remove_none_values(filters))
def fetch_program_course_enrollments_by_student(
user=None,
external_user_key=None,
program_uuids=None,
curriculum_uuids=None,
course_keys=None,
program_enrollment_statuses=None,
active_only=False,
inactive_only=False,
realized_only=False,
waiting_only=False,
):
"""
Fetch program-course enrollments for a specific student.
Required arguments (at least one must be provided):
* user (User)
* external_user_key (str)
Optional arguments:
* provided_uuids (iterable[UUID|str])
* curriculum_uuids (iterable[UUID|str])
* course_keys (iterable[CourseKey|str])
* program_enrollment_statuses (iterable[str])
* realized_only (bool)
* waiting_only (bool)
Optional arguments are used as filtersets if they are not None.
At most one of (realized_only, waiting_only) may be provided.
At most one of (active_only, inactive_only) may be provided.
Returns: queryset[ProgramCourseEnrollment]
"""
if not (user or external_user_key):
raise ValueError(_STUDENT_ARG_ERROR_MESSAGE)
if active_only and inactive_only:
raise ValueError(
_REALIZED_FILTER_ERROR_TEMPLATE.format("active_only", "inactive_only")
)
if realized_only and waiting_only:
raise ValueError(
_REALIZED_FILTER_ERROR_TEMPLATE.format("realized_only", "waiting_only")
)
filters = {
"program_enrollment__user": user,
"program_enrollment__external_user_key": external_user_key,
"program_enrollment__program_uuid__in": program_uuids,
"program_enrollment__curriculum_uuid__in": curriculum_uuids,
"course_key__in": course_keys,
"program_enrollment__status__in": program_enrollment_statuses,
}
if active_only:
filters["status"] = "active"
if inactive_only:
filters["status"] = "inactive"
if realized_only:
filters["program_enrollment__user__isnull"] = False
if waiting_only:
filters["program_enrollment__user__isnull"] = True
return ProgramCourseEnrollment.objects.filter(**_remove_none_values(filters))
def _remove_none_values(dictionary):
"""
Return a dictionary where key-value pairs with `None` as the value
are removed.
"""
return {
key: value for key, value in dictionary.items() if value is not None
}

0
lms/djangoapps/program_enrollments/api/tests/__init__.py Normal file
View File

393
lms/djangoapps/program_enrollments/api/tests/test_linking.py Normal file
View File

@@ -0,0 +1,393 @@
"""
Tests for account linking Python API.
"""
from __future__ import absolute_import, unicode_literals
from uuid import uuid4
from django.test import TestCase
from edx_django_utils.cache import RequestCache
from opaque_keys.edx.keys import CourseKey
from testfixtures import LogCapture
from lms.djangoapps.program_enrollments.tests.factories import ProgramCourseEnrollmentFactory, ProgramEnrollmentFactory
from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
from student.tests.factories import UserFactory
from ..linking import (
COURSE_ENROLLMENT_ERR_TEMPLATE,
NO_LMS_USER_TEMPLATE,
NO_PROGRAM_ENROLLMENT_TEMPLATE,
link_program_enrollments_to_lms_users,
user_already_linked_message
)
LOG_PATH = 'lms.djangoapps.program_enrollments.api.linking'
class TestLinkProgramEnrollmentsMixin(object):
""" Utility methods and test data for testing linking """
@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 linking 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 linking 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_TEMPLATE.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_TEMPLATE.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 = user_already_linked_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 = user_already_linked_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_TEMPLATE.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,
}
)

427
lms/djangoapps/program_enrollments/api/tests/test_reading.py Normal file
View File

@@ -0,0 +1,427 @@
"""
Tests for account linking Python API.
"""
from __future__ import absolute_import, unicode_literals
from uuid import UUID
import ddt
from django.contrib.auth import get_user_model
from django.test import TestCase
from opaque_keys.edx.keys import CourseKey
from course_modes.models import CourseMode
from lms.djangoapps.program_enrollments.constants import ProgramCourseEnrollmentStatuses as PCEStatuses
from lms.djangoapps.program_enrollments.constants import ProgramEnrollmentStatuses as PEStatuses
from lms.djangoapps.program_enrollments.models import ProgramEnrollment
from lms.djangoapps.program_enrollments.tests.factories import ProgramCourseEnrollmentFactory, ProgramEnrollmentFactory
from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
from student.tests.factories import CourseEnrollmentFactory, UserFactory
from ..reading import (
fetch_program_course_enrollments,
fetch_program_course_enrollments_by_student,
fetch_program_enrollments,
fetch_program_enrollments_by_student,
get_program_course_enrollment,
get_program_enrollment
)
User = get_user_model()
@ddt.ddt
class ProgramEnrollmentReadingTests(TestCase):
"""
Tests for program enrollment reading functions.
"""
program_uuid_x = UUID('dddddddd-5f48-493d-9410-84e1d36c657f')
program_uuid_y = UUID('eeeeeeee-f803-43f6-bbf3-5ae15d393649')
program_uuid_z = UUID('ffffffff-89eb-43df-a6b9-c144e7204fd7') # No enrollments
curriculum_uuid_a = UUID('aaaaaaaa-bd26-43d0-94b8-b0063858210b')
curriculum_uuid_b = UUID('bbbbbbbb-145f-43db-ad05-f9ad65eec285')
curriculum_uuid_c = UUID('cccccccc-4577-4559-85f0-4a83e8160a4d')
course_key_p = CourseKey.from_string('course-v1:TestX+ProEnroll+P')
course_key_q = CourseKey.from_string('course-v1:TestX+ProEnroll+Q')
course_key_r = CourseKey.from_string('course-v1:TestX+ProEnroll+R')
username_0 = 'user-0'
username_1 = 'user-1'
username_2 = 'user-2'
username_3 = 'user-3'
username_4 = 'user-4'
ext_3 = 'student-3'
ext_4 = 'student-4'
ext_5 = 'student-5'
ext_6 = 'student-6'
@classmethod
def setUpTestData(cls):
super(ProgramEnrollmentReadingTests, cls).setUpTestData()
cls.user_0 = UserFactory(username=cls.username_0) # No enrollments
cls.user_1 = UserFactory(username=cls.username_1)
cls.user_2 = UserFactory(username=cls.username_2)
cls.user_3 = UserFactory(username=cls.username_3)
cls.user_4 = UserFactory(username=cls.username_4)
CourseOverviewFactory(id=cls.course_key_p)
CourseOverviewFactory(id=cls.course_key_q)
CourseOverviewFactory(id=cls.course_key_r)
enrollment_test_data = [ # ID
(cls.user_1, None, cls.program_uuid_x, cls.curriculum_uuid_a, PEStatuses.ENROLLED), # 1
(cls.user_2, None, cls.program_uuid_x, cls.curriculum_uuid_a, PEStatuses.PENDING), # 2
(cls.user_3, cls.ext_3, cls.program_uuid_x, cls.curriculum_uuid_b, PEStatuses.ENROLLED), # 3
(cls.user_4, cls.ext_4, cls.program_uuid_x, cls.curriculum_uuid_b, PEStatuses.PENDING), # 4
(None, cls.ext_5, cls.program_uuid_x, cls.curriculum_uuid_b, PEStatuses.SUSPENDED), # 5
(None, cls.ext_6, cls.program_uuid_y, cls.curriculum_uuid_c, PEStatuses.CANCELED), # 6
(cls.user_3, cls.ext_3, cls.program_uuid_y, cls.curriculum_uuid_c, PEStatuses.CANCELED), # 7
(None, cls.ext_4, cls.program_uuid_y, cls.curriculum_uuid_c, PEStatuses.ENROLLED), # 8
(cls.user_1, None, cls.program_uuid_x, cls.curriculum_uuid_b, PEStatuses.SUSPENDED), # 9
]
for user, external_user_key, program_uuid, curriculum_uuid, status in enrollment_test_data:
ProgramEnrollmentFactory(
user=user,
external_user_key=external_user_key,
program_uuid=program_uuid,
curriculum_uuid=curriculum_uuid,
status=status,
)
course_enrollment_test_data = [ # ID
(1, cls.course_key_p, PCEStatuses.ACTIVE), # 1
(1, cls.course_key_q, PCEStatuses.ACTIVE), # 2
(9, cls.course_key_r, PCEStatuses.ACTIVE), # 3
(2, cls.course_key_p, PCEStatuses.INACTIVE), # 4
(3, cls.course_key_p, PCEStatuses.ACTIVE), # 5
(5, cls.course_key_p, PCEStatuses.INACTIVE), # 6
(8, cls.course_key_p, PCEStatuses.ACTIVE), # 7
(8, cls.course_key_q, PCEStatuses.INACTIVE), # 8
(2, cls.course_key_r, PCEStatuses.INACTIVE), # 9
(6, cls.course_key_r, PCEStatuses.INACTIVE), # 10
(8, cls.course_key_r, PCEStatuses.ACTIVE), # 11
(7, cls.course_key_q, PCEStatuses.ACTIVE), # 12
]
for program_enrollment_id, course_key, status in course_enrollment_test_data:
program_enrollment = ProgramEnrollment.objects.get(id=program_enrollment_id)
course_enrollment = (
CourseEnrollmentFactory(
course_id=course_key,
user=program_enrollment.user,
mode=CourseMode.MASTERS,
)
if program_enrollment.user
else None
)
ProgramCourseEnrollmentFactory(
program_enrollment=program_enrollment,
course_enrollment=course_enrollment,
course_key=course_key,
status=status,
)
@ddt.data(
# Realized enrollment, specifying only user.
(program_uuid_x, curriculum_uuid_a, username_1, None, 1),
# Realized enrollment, specifiying both user and external key.
(program_uuid_x, curriculum_uuid_b, username_3, ext_3, 3),
# Realized enrollment, specifiying only external key.
(program_uuid_x, curriculum_uuid_b, None, ext_4, 4),
# Waiting enrollment, specifying external key
(program_uuid_x, curriculum_uuid_b, None, ext_5, 5),
# Specifying no curriculum (because ext_6 only has Program Y
# enrollments in one curriculum, so it's not ambiguous).
(program_uuid_y, None, None, ext_6, 6),
)
@ddt.unpack
def test_get_program_enrollment(
self,
program_uuid,
curriculum_uuid,
username,
external_user_key,
expected_enrollment_id,
):
user = User.objects.get(username=username) if username else None
actual_enrollment = get_program_enrollment(
program_uuid=program_uuid,
curriculum_uuid=curriculum_uuid,
user=user,
external_user_key=external_user_key,
)
assert actual_enrollment.id == expected_enrollment_id
@ddt.data(
# Realized enrollment, specifying only user.
(program_uuid_x, None, course_key_p, username_1, None, 1),
# Realized enrollment, specifiying both user and external key.
(program_uuid_x, None, course_key_p, username_3, ext_3, 5),
# Realized enrollment, specifiying only external key.
(program_uuid_y, None, course_key_p, None, ext_4, 7),
# Waiting enrollment, specifying external key
(program_uuid_x, None, course_key_p, None, ext_5, 6),
# We can specify curriculum, but it shouldn't affect anything,
# because each user-course pairing can only have one
# program-course enrollment.
(program_uuid_y, curriculum_uuid_c, course_key_r, None, ext_6, 10),
)
@ddt.unpack
def test_get_program_course_enrollment(
self,
program_uuid,
curriculum_uuid,
course_key,
username,
external_user_key,
expected_enrollment_id,
):
user = User.objects.get(username=username) if username else None
actual_enrollment = get_program_course_enrollment(
program_uuid=program_uuid,
curriculum_uuid=curriculum_uuid,
course_key=course_key,
user=user,
external_user_key=external_user_key,
)
assert actual_enrollment.id == expected_enrollment_id
@ddt.data(
# Program with no enrollments
(
{'program_uuid': program_uuid_z},
set(),
),
# Curriculum & status filters
(
{
'program_uuid': program_uuid_x,
'curriculum_uuids': {curriculum_uuid_a, curriculum_uuid_c},
'program_enrollment_statuses': {PEStatuses.PENDING, PEStatuses.CANCELED},
},
{2},
),
# User & external key filters
(
{
'program_uuid': program_uuid_x,
'usernames': {username_1, username_2, username_3, username_4},
'external_user_keys': {ext_3, ext_4, ext_5}
},
{3, 4},
),
# Realized-only filter
(
{'program_uuid': program_uuid_x, 'realized_only': True},
{1, 2, 3, 4, 9},
),
# Waiting-only filter
(
{'program_uuid': program_uuid_x, 'waiting_only': True},
{5},
),
)
@ddt.unpack
def test_fetch_program_enrollments(self, kwargs, expected_enrollment_ids):
kwargs = self._usernames_to_users(kwargs)
actual_enrollments = fetch_program_enrollments(**kwargs)
actual_enrollment_ids = {enrollment.id for enrollment in actual_enrollments}
assert actual_enrollment_ids == expected_enrollment_ids
@ddt.data(
# Program with no enrollments
(
{'program_uuid': program_uuid_z, 'course_key': course_key_p},
set(),
),
# Curriculum, status, active-only filters
(
{
'program_uuid': program_uuid_x,
'course_key': course_key_p,
'curriculum_uuids': {curriculum_uuid_a, curriculum_uuid_c},
'program_enrollment_statuses': {PEStatuses.ENROLLED},
'active_only': True,
},
{1},
),
# User and external key filters
(
{
'program_uuid': program_uuid_x,
'course_key': course_key_p,
'usernames': {username_2, username_3},
'external_user_keys': {ext_3, ext_5}
},
{5},
),
# Realized-only filter
(
{
'program_uuid': program_uuid_x,
'course_key': course_key_p,
'realized_only': True,
},
{1, 4, 5},
),
# Waiting-only and inactive-only filters
(
{
'program_uuid': program_uuid_y,
'course_key': course_key_r,
'waiting_only': True,
'inactive_only': True,
},
{10},
),
)
@ddt.unpack
def test_fetch_program_course_enrollments(self, kwargs, expected_enrollment_ids):
kwargs = self._usernames_to_users(kwargs)
actual_enrollments = fetch_program_course_enrollments(**kwargs)
actual_enrollment_ids = {enrollment.id for enrollment in actual_enrollments}
assert actual_enrollment_ids == expected_enrollment_ids
@ddt.data(
# User with no enrollments
(
{'username': username_0},
set(),
),
# Filters
(
{
'username': username_3,
'external_user_key': ext_3,
'program_uuids': {program_uuid_x},
'curriculum_uuids': {curriculum_uuid_b, curriculum_uuid_c},
'program_enrollment_statuses': {PEStatuses.ENROLLED, PEStatuses.CANCELED},
},
{3},
),
# More filters
(
{
'username': username_3,
'external_user_key': ext_3,
'program_uuids': {program_uuid_x, program_uuid_y},
'curriculum_uuids': {curriculum_uuid_b, curriculum_uuid_c},
'program_enrollment_statuses': {PEStatuses.SUSPENDED, PEStatuses.CANCELED},
},
{7},
),
# Realized-only filter
(
{'external_user_key': ext_4, 'realized_only': True},
{4},
),
# Waiting-only filter
(
{'external_user_key': ext_4, 'waiting_only': True},
{8},
),
)
@ddt.unpack
def test_fetch_program_enrollments_by_student(self, kwargs, expected_enrollment_ids):
kwargs = self._username_to_user(kwargs)
actual_enrollments = fetch_program_enrollments_by_student(**kwargs)
actual_enrollment_ids = {enrollment.id for enrollment in actual_enrollments}
assert actual_enrollment_ids == expected_enrollment_ids
@ddt.data(
# User with no program enrollments
(
{'username': username_0},
set(),
),
# Course keys and active-only filters
(
{
'external_user_key': ext_4,
'course_keys': {course_key_p, course_key_q},
'active_only': True,
},
{7},
),
# Curriculum filter
(
{'username': username_3, 'curriculum_uuids': {curriculum_uuid_b}},
{5},
),
# Program filter
(
{'username': username_3, 'program_uuids': {program_uuid_y}},
{12},
),
# Realized-only filter
(
{'external_user_key': ext_4, 'realized_only': True},
set(),
),
# Waiting-only and inactive-only filter
(
{
'external_user_key': ext_4,
'waiting_only': True,
'inactive_only': True,
},
{8},
),
)
@ddt.unpack
def test_fetch_program_course_enrollments_by_student(self, kwargs, expected_enrollment_ids):
kwargs = self._username_to_user(kwargs)
actual_enrollments = fetch_program_course_enrollments_by_student(**kwargs)
actual_enrollment_ids = {enrollment.id for enrollment in actual_enrollments}
assert actual_enrollment_ids == expected_enrollment_ids
@staticmethod
def _username_to_user(dictionary):
"""
We can't access the user instances when building `ddt.data`,
so return a dict with the username swapped out for the user themself.
"""
result = dictionary.copy()
if 'username' in result:
result['user'] = User.objects.get(username=result['username'])
del result['username']
return result
@staticmethod
def _usernames_to_users(dictionary):
"""
We can't access the user instances when building `ddt.data`,
so return a dict with the usernames swapped out for the users themselves.
"""
result = dictionary.copy()
if 'usernames' in result:
result['users'] = set(
User.objects.filter(username__in=result['usernames'])
)
del result['usernames']
return result