Merge pull request #20702 from edx/bom/enrollment-readme

Enrollments README and refactor
This commit is contained in:
Nimisha Asthagiri
2019-05-28 16:58:27 -04:00
committed by GitHub
33 changed files with 137 additions and 87 deletions

View File

@@ -0,0 +1,16 @@
Status: Maintenance
Responsibilities
================
The enrollments app provides basic CRUD functionality and APIs for managing Course-Run enrollments.
Enrollments in Programs is managed by the ``lms/djangoapps/program_enrollments\`` app.
Direction: Keep
===============
Glossary
========
More Documentation
==================

View File

@@ -0,0 +1,3 @@
"""
Enrollment API helpers and settings
"""

View File

@@ -0,0 +1,494 @@
"""
Enrollment API for creating, updating, and deleting enrollments. Also provides access to enrollment information at a
course level, such as available course modes.
"""
from __future__ import absolute_import
import importlib
import logging
from django.conf import settings
from django.core.cache import cache
from opaque_keys.edx.keys import CourseKey
from course_modes.models import CourseMode
from openedx.core.djangoapps.enrollments import errors
log = logging.getLogger(__name__)
DEFAULT_DATA_API = 'openedx.core.djangoapps.enrollments.data'
def get_enrollments(user_id, include_inactive=False):
"""Retrieves all the courses a user is enrolled in.
Takes a user and retrieves all relative enrollments. Includes information regarding how the user is enrolled
in the the course.
Args:
user_id (str): The username of the user we want to retrieve course enrollment information for.
include_inactive (bool): Determines whether inactive enrollments will be included
Returns:
A list of enrollment information for the given user.
Examples:
>>> get_enrollments("Bob")
[
{
"created": "2014-10-20T20:18:00Z",
"mode": "honor",
"is_active": True,
"user": "Bob",
"course_details": {
"course_id": "edX/DemoX/2014T2",
"course_name": "edX Demonstration Course",
"enrollment_end": "2014-12-20T20:18:00Z",
"enrollment_start": "2014-10-15T20:18:00Z",
"course_start": "2015-02-03T00:00:00Z",
"course_end": "2015-05-06T00:00:00Z",
"course_modes": [
{
"slug": "honor",
"name": "Honor Code Certificate",
"min_price": 0,
"suggested_prices": "",
"currency": "usd",
"expiration_datetime": null,
"description": null,
"sku": null,
"bulk_sku": null
}
],
"invite_only": False
}
},
{
"created": "2014-10-25T20:18:00Z",
"mode": "verified",
"is_active": True,
"user": "Bob",
"course_details": {
"course_id": "edX/edX-Insider/2014T2",
"course_name": "edX Insider Course",
"enrollment_end": "2014-12-20T20:18:00Z",
"enrollment_start": "2014-10-15T20:18:00Z",
"course_start": "2015-02-03T00:00:00Z",
"course_end": "2015-05-06T00:00:00Z",
"course_modes": [
{
"slug": "honor",
"name": "Honor Code Certificate",
"min_price": 0,
"suggested_prices": "",
"currency": "usd",
"expiration_datetime": null,
"description": null,
"sku": null,
"bulk_sku": null
}
],
"invite_only": True
}
}
]
"""
return _data_api().get_course_enrollments(user_id, include_inactive)
def get_enrollment(user_id, course_id):
"""Retrieves all enrollment information for the user in respect to a specific course.
Gets all the course enrollment information specific to a user in a course.
Args:
user_id (str): The user to get course enrollment information for.
course_id (str): The course to get enrollment information for.
Returns:
A serializable dictionary of the course enrollment.
Example:
>>> get_enrollment("Bob", "edX/DemoX/2014T2")
{
"created": "2014-10-20T20:18:00Z",
"mode": "honor",
"is_active": True,
"user": "Bob",
"course_details": {
"course_id": "edX/DemoX/2014T2",
"course_name": "edX Demonstration Course",
"enrollment_end": "2014-12-20T20:18:00Z",
"enrollment_start": "2014-10-15T20:18:00Z",
"course_start": "2015-02-03T00:00:00Z",
"course_end": "2015-05-06T00:00:00Z",
"course_modes": [
{
"slug": "honor",
"name": "Honor Code Certificate",
"min_price": 0,
"suggested_prices": "",
"currency": "usd",
"expiration_datetime": null,
"description": null,
"sku": null,
"bulk_sku": null
}
],
"invite_only": False
}
}
"""
return _data_api().get_course_enrollment(user_id, course_id)
def add_enrollment(user_id, course_id, mode=None, is_active=True, enrollment_attributes=None):
"""Enrolls a user in a course.
Enrolls a user in a course. If the mode is not specified, this will default to `CourseMode.DEFAULT_MODE_SLUG`.
Arguments:
user_id (str): The user to enroll.
course_id (str): The course to enroll the user in.
mode (str): Optional argument for the type of enrollment to create. Ex. 'audit', 'honor', 'verified',
'professional'. If not specified, this defaults to the default course mode.
is_active (boolean): Optional argument for making the new enrollment inactive. If not specified, is_active
defaults to True.
enrollment_attributes (list): Attributes to be set the enrollment.
Returns:
A serializable dictionary of the new course enrollment.
Example:
>>> add_enrollment("Bob", "edX/DemoX/2014T2", mode="audit")
{
"created": "2014-10-20T20:18:00Z",
"mode": "audit",
"is_active": True,
"user": "Bob",
"course_details": {
"course_id": "edX/DemoX/2014T2",
"course_name": "edX Demonstration Course",
"enrollment_end": "2014-12-20T20:18:00Z",
"enrollment_start": "2014-10-15T20:18:00Z",
"course_start": "2015-02-03T00:00:00Z",
"course_end": "2015-05-06T00:00:00Z",
"course_modes": [
{
"slug": "audit",
"name": "Audit",
"min_price": 0,
"suggested_prices": "",
"currency": "usd",
"expiration_datetime": null,
"description": null,
"sku": null,
"bulk_sku": null
}
],
"invite_only": False
}
}
"""
if mode is None:
mode = _default_course_mode(course_id)
validate_course_mode(course_id, mode, is_active=is_active)
enrollment = _data_api().create_course_enrollment(user_id, course_id, mode, is_active)
if enrollment_attributes is not None:
set_enrollment_attributes(user_id, course_id, enrollment_attributes)
return enrollment
def update_enrollment(user_id, course_id, mode=None, is_active=None, enrollment_attributes=None, include_expired=False):
"""Updates the course mode for the enrolled user.
Update a course enrollment for the given user and course.
Arguments:
user_id (str): The user associated with the updated enrollment.
course_id (str): The course associated with the updated enrollment.
Keyword Arguments:
mode (str): The new course mode for this enrollment.
is_active (bool): Sets whether the enrollment is active or not.
enrollment_attributes (list): Attributes to be set the enrollment.
include_expired (bool): Boolean denoting whether expired course modes should be included.
Returns:
A serializable dictionary representing the updated enrollment.
Example:
>>> update_enrollment("Bob", "edX/DemoX/2014T2", "honor")
{
"created": "2014-10-20T20:18:00Z",
"mode": "honor",
"is_active": True,
"user": "Bob",
"course_details": {
"course_id": "edX/DemoX/2014T2",
"course_name": "edX Demonstration Course",
"enrollment_end": "2014-12-20T20:18:00Z",
"enrollment_start": "2014-10-15T20:18:00Z",
"course_start": "2015-02-03T00:00:00Z",
"course_end": "2015-05-06T00:00:00Z",
"course_modes": [
{
"slug": "honor",
"name": "Honor Code Certificate",
"min_price": 0,
"suggested_prices": "",
"currency": "usd",
"expiration_datetime": null,
"description": null,
"sku": null,
"bulk_sku": null
}
],
"invite_only": False
}
}
"""
log.info(u'Starting Update Enrollment process for user {user} in course {course} to mode {mode}'.format(
user=user_id,
course=course_id,
mode=mode,
))
if mode is not None:
validate_course_mode(course_id, mode, is_active=is_active, include_expired=include_expired)
enrollment = _data_api().update_course_enrollment(user_id, course_id, mode=mode, is_active=is_active)
if enrollment is None:
msg = u"Course Enrollment not found for user {user} in course {course}".format(user=user_id, course=course_id)
log.warn(msg)
raise errors.EnrollmentNotFoundError(msg)
else:
if enrollment_attributes is not None:
set_enrollment_attributes(user_id, course_id, enrollment_attributes)
log.info(u'Course Enrollment updated for user {user} in course {course} to mode {mode}'.format(
user=user_id,
course=course_id,
mode=mode
))
return enrollment
def get_course_enrollment_details(course_id, include_expired=False):
"""Get the course modes for course. Also get enrollment start and end date, invite only, etc.
Given a course_id, return a serializable dictionary of properties describing course enrollment information.
Args:
course_id (str): The Course to get enrollment information for.
include_expired (bool): Boolean denoting whether expired course modes
should be included in the returned JSON data.
Returns:
A serializable dictionary of course enrollment information.
Example:
>>> get_course_enrollment_details("edX/DemoX/2014T2")
{
"course_id": "edX/DemoX/2014T2",
"course_name": "edX Demonstration Course",
"enrollment_end": "2014-12-20T20:18:00Z",
"enrollment_start": "2014-10-15T20:18:00Z",
"course_start": "2015-02-03T00:00:00Z",
"course_end": "2015-05-06T00:00:00Z",
"course_modes": [
{
"slug": "honor",
"name": "Honor Code Certificate",
"min_price": 0,
"suggested_prices": "",
"currency": "usd",
"expiration_datetime": null,
"description": null,
"sku": null,
"bulk_sku": null
}
],
"invite_only": False
}
"""
cache_key = u'enrollment.course.details.{course_id}.{include_expired}'.format(
course_id=course_id,
include_expired=include_expired
)
cached_enrollment_data = None
try:
cached_enrollment_data = cache.get(cache_key)
except Exception: # pylint: disable=broad-except
# The cache backend could raise an exception (for example, memcache keys that contain spaces)
log.exception(u"Error occurred while retrieving course enrollment details from the cache")
if cached_enrollment_data:
log.info(u"Get enrollment data for course %s (cached)", course_id)
return cached_enrollment_data
course_enrollment_details = _data_api().get_course_enrollment_info(course_id, include_expired)
try:
cache_time_out = getattr(settings, 'ENROLLMENT_COURSE_DETAILS_CACHE_TIMEOUT', 60)
cache.set(cache_key, course_enrollment_details, cache_time_out)
except Exception:
# Catch any unexpected errors during caching.
log.exception(u"Error occurred while caching course enrollment details for course %s", course_id)
raise errors.CourseEnrollmentError(u"An unexpected error occurred while retrieving course enrollment details.")
log.info(u"Get enrollment data for course %s", course_id)
return course_enrollment_details
def set_enrollment_attributes(user_id, course_id, attributes):
"""Set enrollment attributes for the enrollment of given user in the
course provided.
Args:
course_id (str): The Course to set enrollment attributes for.
user_id (str): The User to set enrollment attributes for.
attributes (list): Attributes to be set.
Example:
>>>set_enrollment_attributes(
"Bob",
"course-v1-edX-DemoX-1T2015",
[
{
"namespace": "credit",
"name": "provider_id",
"value": "hogwarts",
},
]
)
"""
_data_api().add_or_update_enrollment_attr(user_id, course_id, attributes)
def get_enrollment_attributes(user_id, course_id):
"""Retrieve enrollment attributes for given user for provided course.
Args:
user_id: The User to get enrollment attributes for
course_id (str): The Course to get enrollment attributes for.
Example:
>>>get_enrollment_attributes("Bob", "course-v1-edX-DemoX-1T2015")
[
{
"namespace": "credit",
"name": "provider_id",
"value": "hogwarts",
},
]
Returns: list
"""
return _data_api().get_enrollment_attributes(user_id, course_id)
def _default_course_mode(course_id):
"""Return the default enrollment for a course.
Special case the default enrollment to return if nothing else is found.
Arguments:
course_id (str): The course to check against for available course modes.
Returns:
str
"""
course_modes = CourseMode.modes_for_course(CourseKey.from_string(course_id))
available_modes = [m.slug for m in course_modes]
if CourseMode.DEFAULT_MODE_SLUG in available_modes:
return CourseMode.DEFAULT_MODE_SLUG
elif 'audit' in available_modes:
return 'audit'
elif 'honor' in available_modes:
return 'honor'
return CourseMode.DEFAULT_MODE_SLUG
def validate_course_mode(course_id, mode, is_active=None, include_expired=False):
"""Checks to see if the specified course mode is valid for the course.
If the requested course mode is not available for the course, raise an error with corresponding
course enrollment information.
Arguments:
course_id (str): The course to check against for available course modes.
mode (str): The slug for the course mode specified in the enrollment.
Keyword Arguments:
is_active (bool): Whether the enrollment is to be activated or deactivated.
include_expired (bool): Boolean denoting whether expired course modes should be included.
Returns:
None
Raises:
CourseModeNotFound: raised if the course mode is not found.
"""
# If the client has requested an enrollment deactivation, we want to include expired modes
# in the set of available modes. This allows us to unenroll users from expired modes.
# If include_expired is set as True we should not redetermine its value.
if not include_expired:
include_expired = not is_active if is_active is not None else False
course_enrollment_info = _data_api().get_course_enrollment_info(course_id, include_expired=include_expired)
course_modes = course_enrollment_info["course_modes"]
available_modes = [m['slug'] for m in course_modes]
if mode not in available_modes:
msg = (
u"Specified course mode '{mode}' unavailable for course {course_id}. "
u"Available modes were: {available}"
).format(
mode=mode,
course_id=course_id,
available=", ".join(available_modes)
)
log.warn(msg)
raise errors.CourseModeNotFoundError(msg, course_enrollment_info)
def unenroll_user_from_all_courses(user_id):
"""
Unenrolls a specified user from all of the courses they are currently enrolled in.
:param user_id: The id of the user being unenrolled.
:return: The IDs of all of the organizations from which the learner was unenrolled.
"""
return _data_api().unenroll_user_from_all_courses(user_id)
def get_user_roles(user_id):
"""
Returns a list of all roles that this user has.
:param user_id: The id of the selected user.
:return: All roles for all courses that this user has.
"""
return _data_api().get_user_roles(user_id)
def _data_api():
"""Returns a Data API.
This relies on Django settings to find the appropriate data API.
"""
# We retrieve the settings in-line here (rather than using the
# top-level constant), so that @override_settings will work
# in the test suite.
api_path = getattr(settings, "ENROLLMENT_DATA_API", DEFAULT_DATA_API)
try:
return importlib.import_module(api_path)
except (ImportError, ValueError):
log.exception(u"Could not load module at '{path}'".format(path=api_path))
raise errors.EnrollmentApiLoadError(api_path)

View File

@@ -0,0 +1,356 @@
"""
Data Aggregation Layer of the Enrollment API. Collects all enrollment specific data into a single
source to be used throughout the API.
"""
from __future__ import absolute_import
import logging
from django.contrib.auth.models import User
from django.db import transaction
from opaque_keys.edx.keys import CourseKey
from six import text_type
from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
from openedx.core.djangoapps.enrollments.errors import (
CourseEnrollmentClosedError,
CourseEnrollmentExistsError,
CourseEnrollmentFullError,
InvalidEnrollmentAttribute,
UserNotFoundError
)
from openedx.core.djangoapps.enrollments.serializers import CourseEnrollmentSerializer, CourseSerializer
from openedx.core.lib.exceptions import CourseNotFoundError
from student.models import (
AlreadyEnrolledError,
CourseEnrollment,
CourseEnrollmentAttribute,
CourseFullError,
EnrollmentClosedError,
NonExistentCourseError
)
from student.roles import RoleCache
log = logging.getLogger(__name__)
def get_course_enrollments(user_id, include_inactive=False):
"""Retrieve a list representing all aggregated data for a user's course enrollments.
Construct a representation of all course enrollment data for a specific user.
Args:
user_id (str): The name of the user to retrieve course enrollment information for.
include_inactive (bool): Determines whether inactive enrollments will be included
Returns:
A serializable list of dictionaries of all aggregated enrollment data for a user.
"""
qset = CourseEnrollment.objects.filter(
user__username=user_id,
).order_by('created')
if not include_inactive:
qset = qset.filter(is_active=True)
enrollments = CourseEnrollmentSerializer(qset, many=True).data
# Find deleted courses and filter them out of the results
deleted = []
valid = []
for enrollment in enrollments:
if enrollment.get("course_details") is not None:
valid.append(enrollment)
else:
deleted.append(enrollment)
if deleted:
log.warning(
(
u"Course enrollments for user %s reference "
u"courses that do not exist (this can occur if a course is deleted)."
), user_id,
)
return valid
def get_course_enrollment(username, course_id):
"""Retrieve an object representing all aggregated data for a user's course enrollment.
Get the course enrollment information for a specific user and course.
Args:
username (str): The name of the user to retrieve course enrollment information for.
course_id (str): The course to retrieve course enrollment information for.
Returns:
A serializable dictionary representing the course enrollment.
"""
course_key = CourseKey.from_string(course_id)
try:
enrollment = CourseEnrollment.objects.get(
user__username=username, course_id=course_key
)
return CourseEnrollmentSerializer(enrollment).data
except CourseEnrollment.DoesNotExist:
return None
def get_user_enrollments(course_key):
"""Based on the course id, return all user enrollments in the course
Args:
course_key (CourseKey): Identifier of the course
from which to retrieve enrollments.
Returns:
A course's user enrollments as a queryset
Raises:
CourseEnrollment.DoesNotExist
"""
return CourseEnrollment.objects.filter(
course_id=course_key,
is_active=True
).order_by('created')
def create_course_enrollment(username, course_id, mode, is_active):
"""Create a new course enrollment for the given user.
Creates a new course enrollment for the specified user username.
Args:
username (str): The name of the user to create a new course enrollment for.
course_id (str): The course to create the course enrollment for.
mode (str): (Optional) The mode for the new enrollment.
is_active (boolean): (Optional) Determines if the enrollment is active.
Returns:
A serializable dictionary representing the new course enrollment.
Raises:
CourseNotFoundError
CourseEnrollmentFullError
EnrollmentClosedError
CourseEnrollmentExistsError
"""
course_key = CourseKey.from_string(course_id)
try:
user = User.objects.get(username=username)
except User.DoesNotExist:
msg = u"Not user with username '{username}' found.".format(username=username)
log.warn(msg)
raise UserNotFoundError(msg)
try:
enrollment = CourseEnrollment.enroll(user, course_key, check_access=True)
return _update_enrollment(enrollment, is_active=is_active, mode=mode)
except NonExistentCourseError as err:
raise CourseNotFoundError(text_type(err))
except EnrollmentClosedError as err:
raise CourseEnrollmentClosedError(text_type(err))
except CourseFullError as err:
raise CourseEnrollmentFullError(text_type(err))
except AlreadyEnrolledError as err:
enrollment = get_course_enrollment(username, course_id)
raise CourseEnrollmentExistsError(text_type(err), enrollment)
def update_course_enrollment(username, course_id, mode=None, is_active=None):
"""Modify a course enrollment for a user.
Allows updates to a specific course enrollment.
Args:
username (str): The name of the user to retrieve course enrollment information for.
course_id (str): The course to retrieve course enrollment information for.
mode (str): (Optional) If specified, modify the mode for this enrollment.
is_active (boolean): (Optional) Determines if the enrollment is active.
Returns:
A serializable dictionary representing the modified course enrollment.
"""
course_key = CourseKey.from_string(course_id)
try:
user = User.objects.get(username=username)
except User.DoesNotExist:
msg = u"Not user with username '{username}' found.".format(username=username)
log.warn(msg)
raise UserNotFoundError(msg)
try:
enrollment = CourseEnrollment.objects.get(user=user, course_id=course_key)
return _update_enrollment(enrollment, is_active=is_active, mode=mode)
except CourseEnrollment.DoesNotExist:
return None
def add_or_update_enrollment_attr(user_id, course_id, attributes):
"""Set enrollment attributes for the enrollment of given user in the
course provided.
Args:
course_id (str): The Course to set enrollment attributes for.
user_id (str): The User to set enrollment attributes for.
attributes (list): Attributes to be set.
Example:
>>>add_or_update_enrollment_attr(
"Bob",
"course-v1-edX-DemoX-1T2015",
[
{
"namespace": "credit",
"name": "provider_id",
"value": "hogwarts",
},
]
)
"""
course_key = CourseKey.from_string(course_id)
user = _get_user(user_id)
enrollment = CourseEnrollment.get_enrollment(user, course_key)
if not _invalid_attribute(attributes) and enrollment is not None:
CourseEnrollmentAttribute.add_enrollment_attr(enrollment, attributes)
def get_enrollment_attributes(user_id, course_id):
"""Retrieve enrollment attributes for given user for provided course.
Args:
user_id: The User to get enrollment attributes for
course_id (str): The Course to get enrollment attributes for.
Example:
>>>get_enrollment_attributes("Bob", "course-v1-edX-DemoX-1T2015")
[
{
"namespace": "credit",
"name": "provider_id",
"value": "hogwarts",
},
]
Returns: list
"""
course_key = CourseKey.from_string(course_id)
user = _get_user(user_id)
enrollment = CourseEnrollment.get_enrollment(user, course_key)
return CourseEnrollmentAttribute.get_enrollment_attributes(enrollment)
def unenroll_user_from_all_courses(user_id):
"""
Set all of a user's enrollments to inactive.
:param user_id: The user being unenrolled.
:return: A list of all courses from which the user was unenrolled.
"""
user = _get_user(user_id)
enrollments = CourseEnrollment.objects.filter(user=user)
with transaction.atomic():
for enrollment in enrollments:
_update_enrollment(enrollment, is_active=False)
return set([str(enrollment.course_id.org) for enrollment in enrollments])
def _get_user(user_id):
"""Retrieve user with provided user_id
Args:
user_id(str): username of the user for which object is to retrieve
Returns: obj
"""
try:
return User.objects.get(username=user_id)
except User.DoesNotExist:
msg = u"Not user with username '{username}' found.".format(username=user_id)
log.warn(msg)
raise UserNotFoundError(msg)
def _update_enrollment(enrollment, is_active=None, mode=None):
enrollment.update_enrollment(is_active=is_active, mode=mode)
enrollment.save()
return CourseEnrollmentSerializer(enrollment).data
def _invalid_attribute(attributes):
"""Validate enrollment attribute
Args:
attributes(dict): dict of attribute
Return:
list of invalid attributes
"""
invalid_attributes = []
for attribute in attributes:
if "namespace" not in attribute:
msg = u"'namespace' not in enrollment attribute"
log.warn(msg)
invalid_attributes.append("namespace")
raise InvalidEnrollmentAttribute(msg)
if "name" not in attribute:
msg = u"'name' not in enrollment attribute"
log.warn(msg)
invalid_attributes.append("name")
raise InvalidEnrollmentAttribute(msg)
if "value" not in attribute:
msg = u"'value' not in enrollment attribute"
log.warn(msg)
invalid_attributes.append("value")
raise InvalidEnrollmentAttribute(msg)
return invalid_attributes
def get_course_enrollment_info(course_id, include_expired=False):
"""Returns all course enrollment information for the given course.
Based on the course id, return all related course information.
Args:
course_id (str): The course to retrieve enrollment information for.
include_expired (bool): Boolean denoting whether expired course modes
should be included in the returned JSON data.
Returns:
A serializable dictionary representing the course's enrollment information.
Raises:
CourseNotFoundError
"""
course_key = CourseKey.from_string(course_id)
try:
course = CourseOverview.get_from_id(course_key)
except CourseOverview.DoesNotExist:
msg = u"Requested enrollment information for unknown course {course}".format(course=course_id)
log.warning(msg)
raise CourseNotFoundError(msg)
else:
return CourseSerializer(course, include_expired=include_expired).data
def get_user_roles(user_id):
"""
Returns a list of all roles that this user has.
:param user_id: The id of the selected user.
:return: All roles for all courses that this user has.
"""
# pylint: disable=protected-access
user = _get_user(user_id)
if not hasattr(user, '_roles'):
user._roles = RoleCache(user)
role_cache = user._roles
return role_cache._roles

View File

@@ -0,0 +1,53 @@
"""All Error Types pertaining to Enrollment."""
class CourseEnrollmentError(Exception):
"""Generic Course Enrollment Error.
Describes any error that may occur when reading or updating enrollment information for a user or a course.
"""
def __init__(self, msg, data=None):
super(CourseEnrollmentError, self).__init__(msg)
# Corresponding information to help resolve the error.
self.data = data
class UserNotFoundError(CourseEnrollmentError):
pass
class CourseEnrollmentClosedError(CourseEnrollmentError):
pass
class CourseEnrollmentFullError(CourseEnrollmentError):
pass
class CourseEnrollmentExistsError(CourseEnrollmentError):
enrollment = None
def __init__(self, message, enrollment):
super(CourseEnrollmentExistsError, self).__init__(message)
self.enrollment = enrollment
class CourseModeNotFoundError(CourseEnrollmentError):
"""The requested course mode could not be found."""
pass
class EnrollmentNotFoundError(CourseEnrollmentError):
"""The requested enrollment could not be found."""
pass
class EnrollmentApiLoadError(CourseEnrollmentError):
"""The data API could not be loaded."""
pass
class InvalidEnrollmentAttribute(CourseEnrollmentError):
"""Enrollment Attributes could not be validated"""
pass

View File

@@ -0,0 +1,51 @@
"""
Forms for validating user input to the Course Enrollment related views.
"""
from __future__ import absolute_import
from django.core.exceptions import ValidationError
from django.forms import CharField, Form
from opaque_keys import InvalidKeyError
from opaque_keys.edx.keys import CourseKey
from student import forms as student_forms
class CourseEnrollmentsApiListForm(Form):
"""
A form that validates the query string parameters for the CourseEnrollmentsApiListView.
"""
MAX_USERNAME_COUNT = 100
username = CharField(required=False)
course_id = CharField(required=False)
def clean_course_id(self):
"""
Validate and return a course ID.
"""
course_id = self.cleaned_data.get('course_id')
if course_id:
try:
return CourseKey.from_string(course_id)
except InvalidKeyError:
raise ValidationError(u"'{}' is not a valid course id.".format(course_id))
return course_id
def clean_username(self):
"""
Validate a string of comma-separated usernames and return a list of usernames.
"""
usernames_csv_string = self.cleaned_data.get('username')
if usernames_csv_string:
usernames = usernames_csv_string.split(',')
if len(usernames) > self.MAX_USERNAME_COUNT:
raise ValidationError(
u"Too many usernames in a single request - {}. A maximum of {} is allowed".format(
len(usernames),
self.MAX_USERNAME_COUNT,
)
)
for username in usernames:
student_forms.validate_username(username)
return usernames
return usernames_csv_string

View File

@@ -0,0 +1,62 @@
"""
Management command for enrolling a user into a course via the enrollment api
"""
from __future__ import absolute_import
from django.contrib.auth.models import User
from django.core.management.base import BaseCommand
from openedx.core.djangoapps.enrollments.data import CourseEnrollmentExistsError
from openedx.core.djangoapps.enrollments.api import add_enrollment
class Command(BaseCommand):
"""
Enroll a user into a course
"""
help = """
This enrolls a user into a given course
User email and course ID are required.
Mode is optional. It defaults to the default mode (e.g., 'honor', 'audit', etc).
example:
# Enroll a user test@example.com into the demo course
manage.py ... enroll_user_in_course -e test@example.com -c edX/Open_DemoX/edx_demo_course
This command can be run multiple times on the same user+course (i.e. it is idempotent).
"""
def add_arguments(self, parser):
parser.add_argument(
'-e', '--email',
nargs=1,
required=True,
help='Email for user'
)
parser.add_argument(
'-c', '--course',
nargs=1,
required=True,
help='course ID to enroll the user in'
)
parser.add_argument(
'-m', '--mode',
required=False,
default=None,
help='course mode to enroll the user in'
)
def handle(self, *args, **options):
"""
Get and enroll a user in the given course. Mode is optional and defers to the enrollment API for defaults.
"""
email = options['email'][0]
course = options['course'][0]
mode = options['mode']
user = User.objects.get(email=email)
try:
add_enrollment(user.username, course, mode=mode)
except CourseEnrollmentExistsError:
# If the user is already enrolled in the course, do nothing.
pass

View File

@@ -0,0 +1,90 @@
""" Test the change_enrollment command line script."""
from __future__ import absolute_import
import unittest
from uuid import uuid4
import ddt
from django.conf import settings
from django.core.management import call_command
from django.core.management.base import CommandError
from openedx.core.djangoapps.enrollments.api import get_enrollment
from student.tests.factories import UserFactory
from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
import six
from six.moves import range
@ddt.ddt
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
class EnrollManagementCommandTest(SharedModuleStoreTestCase):
"""
Test the enroll_user_in_course management command
"""
@classmethod
def setUpClass(cls):
super(EnrollManagementCommandTest, cls).setUpClass()
cls.course = CourseFactory.create(org='fooX', number='007')
def setUp(self):
super(EnrollManagementCommandTest, self).setUp()
self.course_id = six.text_type(self.course.id)
self.username = 'ralph' + uuid4().hex
self.user_email = self.username + '@example.com'
UserFactory(username=self.username, email=self.user_email)
def test_enroll_user(self):
command_args = [
'--course', self.course_id,
'--email', self.user_email,
]
call_command(
'enroll_user_in_course',
*command_args
)
user_enroll = get_enrollment(self.username, self.course_id)
self.assertTrue(user_enroll['is_active'])
def test_enroll_user_twice(self):
"""
Ensures the command is idempotent.
"""
command_args = [
'--course', self.course_id,
'--email', self.user_email,
]
for _ in range(2):
call_command(
'enroll_user_in_course',
*command_args
)
# Second run does not impact the first run (i.e., the
# user is still enrolled, no exception was raised, etc)
user_enroll = get_enrollment(self.username, self.course_id)
self.assertTrue(user_enroll['is_active'])
@ddt.data(['--email', 'foo'], ['--course', 'bar'], ['--bad-param', 'baz'])
def test_not_enough_args(self, arg):
"""
When the command is missing certain arguments, it should
raise an exception
"""
command_args = arg
with self.assertRaises(CommandError):
call_command(
'enroll_user_in_course',
*command_args
)

View File

@@ -0,0 +1,13 @@
"""
Paginators for the course enrollment related views.
"""
from __future__ import absolute_import
from rest_framework.pagination import CursorPagination
class CourseEnrollmentsApiListPagination(CursorPagination):
"""
Paginator for the Course enrollments list API.
"""
page_size = 100

View File

@@ -0,0 +1,118 @@
"""
Serializers for all Course Enrollment related return objects.
"""
from __future__ import absolute_import
import logging
from rest_framework import serializers
from course_modes.models import CourseMode
from student.models import CourseEnrollment
log = logging.getLogger(__name__)
class StringListField(serializers.CharField):
"""Custom Serializer for turning a comma delimited string into a list.
This field is designed to take a string such as "1,2,3" and turn it into an actual list
[1,2,3]
"""
def field_to_native(self, obj, field_name): # pylint: disable=unused-argument
"""
Serialize the object's class name.
"""
if not obj.suggested_prices:
return []
items = obj.suggested_prices.split(',')
return [int(item) for item in items]
class CourseSerializer(serializers.Serializer): # pylint: disable=abstract-method
"""
Serialize a course descriptor and related information.
"""
course_id = serializers.CharField(source="id")
course_name = serializers.CharField(source="display_name_with_default")
enrollment_start = serializers.DateTimeField(format=None)
enrollment_end = serializers.DateTimeField(format=None)
course_start = serializers.DateTimeField(source="start", format=None)
course_end = serializers.DateTimeField(source="end", format=None)
invite_only = serializers.BooleanField(source="invitation_only")
course_modes = serializers.SerializerMethodField()
def __init__(self, *args, **kwargs):
self.include_expired = kwargs.pop("include_expired", False)
super(CourseSerializer, self).__init__(*args, **kwargs)
def get_course_modes(self, obj):
"""
Retrieve course modes associated with the course.
"""
course_modes = CourseMode.modes_for_course(
obj.id,
include_expired=self.include_expired,
only_selectable=False
)
return [
ModeSerializer(mode).data
for mode in course_modes
]
class CourseEnrollmentSerializer(serializers.ModelSerializer):
"""Serializes CourseEnrollment models
Aggregates all data from the Course Enrollment table, and pulls in the serialization for
the Course Descriptor and course modes, to give a complete representation of course enrollment.
"""
course_details = CourseSerializer(source="course_overview")
user = serializers.SerializerMethodField('get_username')
def get_username(self, model):
"""Retrieves the username from the associated model."""
return model.username
class Meta(object):
model = CourseEnrollment
fields = ('created', 'mode', 'is_active', 'course_details', 'user')
lookup_field = 'username'
class CourseEnrollmentsApiListSerializer(CourseEnrollmentSerializer):
"""
Serializes CourseEnrollment model and returns a subset of fields returned
by the CourseEnrollmentSerializer.
"""
course_id = serializers.CharField(source='course_overview.id')
def __init__(self, *args, **kwargs):
super(CourseEnrollmentsApiListSerializer, self).__init__(*args, **kwargs)
self.fields.pop('course_details')
class Meta(CourseEnrollmentSerializer.Meta):
fields = CourseEnrollmentSerializer.Meta.fields + ('course_id', )
class ModeSerializer(serializers.Serializer): # pylint: disable=abstract-method
"""Serializes a course's 'Mode' tuples
Returns a serialized representation of the modes available for course enrollment. The course
modes models are designed to return a tuple instead of the model object itself. This serializer
does not handle the model object itself, but the tuple.
"""
slug = serializers.CharField(max_length=100)
name = serializers.CharField(max_length=255)
min_price = serializers.IntegerField()
suggested_prices = StringListField(max_length=255)
currency = serializers.CharField(max_length=8)
expiration_datetime = serializers.DateTimeField()
description = serializers.CharField()
sku = serializers.CharField()
bulk_sku = serializers.CharField()

View File

@@ -0,0 +1,138 @@
"""
A Fake Data API for testing purposes.
"""
from __future__ import absolute_import
import copy
import datetime
_DEFAULT_FAKE_MODE = {
"slug": "honor",
"name": "Honor Code Certificate",
"min_price": 0,
"suggested_prices": "",
"currency": "usd",
"expiration_datetime": None,
"description": None
}
_ENROLLMENTS = []
_COURSES = []
_ENROLLMENT_ATTRIBUTES = []
_VERIFIED_MODE_EXPIRED = []
# pylint: disable=unused-argument
def get_course_enrollments(student_id, include_inactive=False):
"""Stubbed out Enrollment data request."""
return _ENROLLMENTS
def get_course_enrollment(student_id, course_id):
"""Stubbed out Enrollment data request."""
return _get_fake_enrollment(student_id, course_id)
def create_course_enrollment(student_id, course_id, mode='honor', is_active=True):
"""Stubbed out Enrollment creation request. """
return add_enrollment(student_id, course_id, mode=mode, is_active=is_active)
def update_course_enrollment(student_id, course_id, mode=None, is_active=None):
"""Stubbed out Enrollment data request."""
enrollment = _get_fake_enrollment(student_id, course_id)
if enrollment and mode is not None:
enrollment['mode'] = mode
if enrollment and is_active is not None:
enrollment['is_active'] = is_active
return enrollment
def get_course_enrollment_info(course_id, include_expired=False):
"""Stubbed out Enrollment data request."""
return _get_fake_course_info(course_id, include_expired)
def _get_fake_enrollment(student_id, course_id):
"""Get an enrollment from the enrollments array."""
for enrollment in _ENROLLMENTS:
if student_id == enrollment['student'] and course_id == enrollment['course']['course_id']:
return enrollment
def _get_fake_course_info(course_id, include_expired=False):
"""Get a course from the courses array."""
# if verified mode is expired and include expired is false
# then remove the verified mode from the course.
for course in _COURSES:
if course_id == course['course_id']:
if course_id in _VERIFIED_MODE_EXPIRED and not include_expired:
course['course_modes'] = [mode for mode in course['course_modes'] if mode['slug'] != 'verified']
return course
def add_enrollment(student_id, course_id, is_active=True, mode='honor'):
"""Append an enrollment to the enrollments array."""
enrollment = {
"created": datetime.datetime.now(),
"mode": mode,
"is_active": is_active,
"course": _get_fake_course_info(course_id),
"student": student_id
}
_ENROLLMENTS.append(enrollment)
return enrollment
# pylint: disable=unused-argument
def add_or_update_enrollment_attr(user_id, course_id, attributes):
"""Add or update enrollment attribute array"""
for attribute in attributes:
_ENROLLMENT_ATTRIBUTES.append({
'namespace': attribute['namespace'],
'name': attribute['name'],
'value': attribute['value']
})
# pylint: disable=unused-argument
def get_enrollment_attributes(user_id, course_id):
"""Retrieve enrollment attribute array"""
return _ENROLLMENT_ATTRIBUTES
def set_expired_mode(course_id):
"""Set course verified mode as expired."""
_VERIFIED_MODE_EXPIRED.append(course_id)
def add_course(course_id, enrollment_start=None, enrollment_end=None, invite_only=False, course_modes=None):
"""Append course to the courses array."""
course_info = {
"course_id": course_id,
"enrollment_end": enrollment_end,
"course_modes": [],
"enrollment_start": enrollment_start,
"invite_only": invite_only,
}
if not course_modes:
course_info['course_modes'].append(_DEFAULT_FAKE_MODE)
else:
for mode in course_modes:
new_mode = copy.deepcopy(_DEFAULT_FAKE_MODE)
new_mode['slug'] = mode
course_info['course_modes'].append(new_mode)
_COURSES.append(course_info)
def reset():
"""Set the enrollments and courses arrays to be empty."""
global _COURSES # pylint: disable=global-statement
_COURSES = []
global _ENROLLMENTS # pylint: disable=global-statement
_ENROLLMENTS = []
global _VERIFIED_MODE_EXPIRED # pylint: disable=global-statement
_VERIFIED_MODE_EXPIRED = []

View File

@@ -0,0 +1,157 @@
[
[
{
"course_id": "e/d/X"
},
[
{
"course_id": "e/d/X",
"is_active": true,
"mode": "honor",
"user": "student1",
"created": "2018-01-01T00:00:01Z"
},
{
"course_id": "e/d/X",
"is_active": true,
"mode": "honor",
"user": "student2",
"created": "2018-01-01T00:00:01Z"
}
]
],
[
{
"course_id": "x/y/Z"
},
[
{
"course_id": "x/y/Z",
"is_active": true,
"mode": "verified",
"user": "staff",
"created": "2018-01-01T00:00:01Z"
},
{
"course_id": "x/y/Z",
"is_active": true,
"mode": "honor",
"user": "student2",
"created": "2018-01-01T00:00:01Z"
},
{
"course_id": "x/y/Z",
"is_active": true,
"mode": "verified",
"user": "student3",
"created": "2018-01-01T00:00:01Z"
}
]
],
[
{
"course_id": "x/y/Z",
"username": "student2,student3"
},
[
{
"course_id": "x/y/Z",
"is_active": true,
"mode": "honor",
"user": "student2",
"created": "2018-01-01T00:00:01Z"
},
{
"course_id": "x/y/Z",
"is_active": true,
"mode": "verified",
"user": "student3",
"created": "2018-01-01T00:00:01Z"
}
]
],
[
{
"course_id": "x/y/Z",
"username": "student1,student2"
},
[
{
"course_id": "x/y/Z",
"is_active": true,
"mode": "honor",
"user": "student2",
"created": "2018-01-01T00:00:01Z"
}
]
],
[
{
"username": "student2,staff"
},
[
{
"course_id": "x/y/Z",
"is_active": true,
"mode": "verified",
"user": "staff",
"created": "2018-01-01T00:00:01Z"
},
{
"course_id": "e/d/X",
"is_active": true,
"mode": "honor",
"user": "student2",
"created": "2018-01-01T00:00:01Z"
},
{
"course_id": "x/y/Z",
"is_active": true,
"mode": "honor",
"user": "student2",
"created": "2018-01-01T00:00:01Z"
}
]
],
[
null,
[
{
"course_id": "e/d/X",
"is_active": true,
"mode": "honor",
"user": "student1",
"created": "2018-01-01T00:00:01Z"
},
{
"course_id": "e/d/X",
"is_active": true,
"mode": "honor",
"user": "student2",
"created": "2018-01-01T00:00:01Z"
},
{
"course_id": "x/y/Z",
"is_active": true,
"mode": "verified",
"user": "student3",
"created": "2018-01-01T00:00:01Z"
},
{
"course_id": "x/y/Z",
"is_active": true,
"mode": "honor",
"user": "student2",
"created": "2018-01-01T00:00:01Z"
},
{
"course_id": "x/y/Z",
"is_active": true,
"mode": "verified",
"user": "staff",
"created": "2018-01-01T00:00:01Z"
}
]
]
]

View File

@@ -0,0 +1,283 @@
"""
Tests for student enrollment.
"""
from __future__ import absolute_import
import unittest
import ddt
import pytest
from django.conf import settings
from django.test.utils import override_settings
from mock import Mock, patch
from course_modes.models import CourseMode
from openedx.core.djangoapps.enrollments import api
from openedx.core.djangoapps.enrollments.errors import (
CourseModeNotFoundError, EnrollmentApiLoadError, EnrollmentNotFoundError,
)
from openedx.core.djangoapps.enrollments.tests import fake_data_api
from openedx.core.djangolib.testing.utils import CacheIsolationTestCase
@ddt.ddt
@override_settings(ENROLLMENT_DATA_API="openedx.core.djangoapps.enrollments.tests.fake_data_api")
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
class EnrollmentTest(CacheIsolationTestCase):
"""
Test student enrollment, especially with different course modes.
"""
USERNAME = "Bob"
COURSE_ID = "some/great/course"
ENABLED_CACHES = ['default']
def setUp(self):
super(EnrollmentTest, self).setUp()
fake_data_api.reset()
@ddt.data(
# Default (no course modes in the database)
# Expect automatically being enrolled as "honor".
([], 'honor'),
# Audit / Verified / Honor
# We should always go to the "choose your course" page.
# We should also be enrolled as "honor" by default.
(['honor', 'verified', 'audit'], 'honor'),
# Check for professional ed happy path.
(['professional'], 'professional'),
(['no-id-professional'], 'no-id-professional')
)
@ddt.unpack
def test_enroll(self, course_modes, mode):
# Add a fake course enrollment information to the fake data API
fake_data_api.add_course(self.COURSE_ID, course_modes=course_modes)
# Enroll in the course and verify the URL we get sent to
result = api.add_enrollment(self.USERNAME, self.COURSE_ID, mode=mode)
self.assertIsNotNone(result)
self.assertEquals(result['student'], self.USERNAME)
self.assertEquals(result['course']['course_id'], self.COURSE_ID)
self.assertEquals(result['mode'], mode)
get_result = api.get_enrollment(self.USERNAME, self.COURSE_ID)
self.assertEquals(result, get_result)
@ddt.data(
([CourseMode.DEFAULT_MODE_SLUG, 'verified', 'credit'], CourseMode.DEFAULT_MODE_SLUG),
(['audit', 'verified', 'credit'], 'audit'),
(['honor', 'verified', 'credit'], 'honor'),
)
@ddt.unpack
def test_enroll_no_mode_success(self, course_modes, expected_mode):
# Add a fake course enrollment information to the fake data API
fake_data_api.add_course(self.COURSE_ID, course_modes=course_modes)
with patch('openedx.core.djangoapps.enrollments.api.CourseMode.modes_for_course') as mock_modes_for_course:
mock_course_modes = [Mock(slug=mode) for mode in course_modes]
mock_modes_for_course.return_value = mock_course_modes
# Enroll in the course and verify the URL we get sent to
result = api.add_enrollment(self.USERNAME, self.COURSE_ID)
self.assertIsNotNone(result)
self.assertEquals(result['student'], self.USERNAME)
self.assertEquals(result['course']['course_id'], self.COURSE_ID)
self.assertEquals(result['mode'], expected_mode)
@ddt.data(
['professional'],
['verified'],
['verified', 'professional'],
)
def test_enroll_no_mode_error(self, course_modes):
# Add a fake course enrollment information to the fake data API
fake_data_api.add_course(self.COURSE_ID, course_modes=course_modes)
# Enroll in the course and verify that we raise CourseModeNotFoundError
with pytest.raises(CourseModeNotFoundError):
api.add_enrollment(self.USERNAME, self.COURSE_ID)
def test_prof_ed_enroll(self):
# Add a fake course enrollment information to the fake data API
fake_data_api.add_course(self.COURSE_ID, course_modes=['professional'])
# Enroll in the course and verify the URL we get sent to
with pytest.raises(CourseModeNotFoundError):
api.add_enrollment(self.USERNAME, self.COURSE_ID, mode='verified')
@ddt.data(
# Default (no course modes in the database)
# Expect that users are automatically enrolled as "honor".
([], 'honor'),
# Audit / Verified / Honor
# We should always go to the "choose your course" page.
# We should also be enrolled as "honor" by default.
(['honor', 'verified', 'audit'], 'honor'),
# Check for professional ed happy path.
(['professional'], 'professional'),
(['no-id-professional'], 'no-id-professional')
)
@ddt.unpack
def test_unenroll(self, course_modes, mode):
# Add a fake course enrollment information to the fake data API
fake_data_api.add_course(self.COURSE_ID, course_modes=course_modes)
# Enroll in the course and verify the URL we get sent to
result = api.add_enrollment(self.USERNAME, self.COURSE_ID, mode=mode)
self.assertIsNotNone(result)
self.assertEquals(result['student'], self.USERNAME)
self.assertEquals(result['course']['course_id'], self.COURSE_ID)
self.assertEquals(result['mode'], mode)
self.assertTrue(result['is_active'])
result = api.update_enrollment(self.USERNAME, self.COURSE_ID, mode=mode, is_active=False)
self.assertIsNotNone(result)
self.assertEquals(result['student'], self.USERNAME)
self.assertEquals(result['course']['course_id'], self.COURSE_ID)
self.assertEquals(result['mode'], mode)
self.assertFalse(result['is_active'])
def test_unenroll_not_enrolled_in_course(self):
# Add a fake course enrollment information to the fake data API
fake_data_api.add_course(self.COURSE_ID, course_modes=['honor'])
with pytest.raises(EnrollmentNotFoundError):
api.update_enrollment(self.USERNAME, self.COURSE_ID, mode='honor', is_active=False)
@ddt.data(
# Simple test of honor and verified.
([
{'course_id': 'the/first/course', 'course_modes': [], 'mode': 'honor'},
{'course_id': 'the/second/course', 'course_modes': ['honor', 'verified'], 'mode': 'verified'}
]),
# No enrollments
([]),
# One Enrollment
([
{'course_id': 'the/third/course', 'course_modes': ['honor', 'verified', 'audit'], 'mode': 'audit'}
]),
)
def test_get_all_enrollments(self, enrollments):
for enrollment in enrollments:
fake_data_api.add_course(enrollment['course_id'], course_modes=enrollment['course_modes'])
api.add_enrollment(self.USERNAME, enrollment['course_id'], enrollment['mode'])
result = api.get_enrollments(self.USERNAME)
self.assertEqual(len(enrollments), len(result))
for result_enrollment in result:
self.assertIn(
result_enrollment['course']['course_id'],
[enrollment['course_id'] for enrollment in enrollments]
)
def test_update_enrollment(self):
# Add fake course enrollment information to the fake data API
fake_data_api.add_course(self.COURSE_ID, course_modes=['honor', 'verified', 'audit'])
# Enroll in the course and verify the URL we get sent to
result = api.add_enrollment(self.USERNAME, self.COURSE_ID, mode='audit')
get_result = api.get_enrollment(self.USERNAME, self.COURSE_ID)
self.assertEquals(result, get_result)
result = api.update_enrollment(self.USERNAME, self.COURSE_ID, mode='honor')
self.assertEquals('honor', result['mode'])
result = api.update_enrollment(self.USERNAME, self.COURSE_ID, mode='verified')
self.assertEquals('verified', result['mode'])
def test_update_enrollment_attributes(self):
# Add fake course enrollment information to the fake data API
fake_data_api.add_course(self.COURSE_ID, course_modes=['honor', 'verified', 'audit', 'credit'])
# Enroll in the course and verify the URL we get sent to
result = api.add_enrollment(self.USERNAME, self.COURSE_ID, mode='audit')
get_result = api.get_enrollment(self.USERNAME, self.COURSE_ID)
self.assertEquals(result, get_result)
enrollment_attributes = [
{
"namespace": "credit",
"name": "provider_id",
"value": "hogwarts",
}
]
result = api.update_enrollment(
self.USERNAME, self.COURSE_ID, mode='credit', enrollment_attributes=enrollment_attributes
)
self.assertEquals('credit', result['mode'])
attributes = api.get_enrollment_attributes(self.USERNAME, self.COURSE_ID)
self.assertEquals(enrollment_attributes[0], attributes[0])
def test_get_course_details(self):
# Add a fake course enrollment information to the fake data API
fake_data_api.add_course(self.COURSE_ID, course_modes=['honor', 'verified', 'audit'])
result = api.get_course_enrollment_details(self.COURSE_ID)
self.assertEquals(result['course_id'], self.COURSE_ID)
self.assertEquals(3, len(result['course_modes']))
@override_settings(ENROLLMENT_DATA_API='foo.bar.biz.baz')
def test_data_api_config_error(self):
# Enroll in the course and verify the URL we get sent to
with pytest.raises(EnrollmentApiLoadError):
api.add_enrollment(self.USERNAME, self.COURSE_ID, mode='audit')
def test_caching(self):
# Add fake course enrollment information to the fake data API
fake_data_api.add_course(self.COURSE_ID, course_modes=['honor', 'verified', 'audit'])
# Hit the fake data API.
details = api.get_course_enrollment_details(self.COURSE_ID)
# Reset the fake data API, should rely on the cache.
fake_data_api.reset()
cached_details = api.get_course_enrollment_details(self.COURSE_ID)
# The data matches
self.assertEqual(len(details['course_modes']), 3)
self.assertEqual(details, cached_details)
def test_update_enrollment_expired_mode_with_error(self):
""" Verify that if verified mode is expired and include expire flag is
false then enrollment cannot be updated. """
self.assert_add_modes_with_enrollment('audit')
# On updating enrollment mode to verified it should the raise the error.
with self.assertRaises(CourseModeNotFoundError):
self.assert_update_enrollment(mode='verified', include_expired=False)
def test_update_enrollment_with_expired_mode(self):
""" Verify that if verified mode is expired then enrollment can be
updated if include_expired flag is true."""
self.assert_add_modes_with_enrollment('audit')
# enrollment in verified mode will work fine with include_expired=True
self.assert_update_enrollment(mode='verified', include_expired=True)
@ddt.data(True, False)
def test_unenroll_with_expired_mode(self, include_expired):
""" Verify that un-enroll will work fine for expired courses whether include_expired
is true or false."""
self.assert_add_modes_with_enrollment('verified')
self.assert_update_enrollment(mode='verified', is_active=False, include_expired=include_expired)
def assert_add_modes_with_enrollment(self, enrollment_mode):
""" Dry method for adding fake course enrollment information to fake
data API and enroll the student in the course. """
fake_data_api.add_course(self.COURSE_ID, course_modes=['honor', 'verified', 'audit'])
result = api.add_enrollment(self.USERNAME, self.COURSE_ID, mode=enrollment_mode)
get_result = api.get_enrollment(self.USERNAME, self.COURSE_ID)
self.assertEquals(result, get_result)
# set the course verify mode as expire.
fake_data_api.set_expired_mode(self.COURSE_ID)
def assert_update_enrollment(self, mode, is_active=True, include_expired=False):
""" Dry method for updating enrollment."""
result = api.update_enrollment(
self.USERNAME, self.COURSE_ID, mode=mode, is_active=is_active, include_expired=include_expired
)
self.assertEquals(mode, result['mode'])
self.assertIsNotNone(result)
self.assertEquals(result['student'], self.USERNAME)
self.assertEquals(result['course']['course_id'], self.COURSE_ID)
self.assertEquals(result['mode'], mode)
if is_active:
self.assertTrue(result['is_active'])
else:
self.assertFalse(result['is_active'])

View File

@@ -0,0 +1,402 @@
"""
Test the Data Aggregation Layer for Course Enrollments.
"""
from __future__ import absolute_import
import datetime
import unittest
import ddt
import pytest
import six
from django.conf import settings
from mock import patch
from pytz import UTC
from six.moves import range
from course_modes.models import CourseMode
from course_modes.tests.factories import CourseModeFactory
from openedx.core.djangoapps.enrollments import data
from openedx.core.djangoapps.enrollments.errors import (
CourseEnrollmentClosedError,
CourseEnrollmentExistsError,
CourseEnrollmentFullError,
UserNotFoundError
)
from openedx.core.djangoapps.enrollments.serializers import CourseEnrollmentSerializer
from openedx.core.lib.exceptions import CourseNotFoundError
from student.models import AlreadyEnrolledError, CourseEnrollment, CourseFullError, EnrollmentClosedError
from student.tests.factories import CourseAccessRoleFactory, UserFactory
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
@ddt.ddt
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
class EnrollmentDataTest(ModuleStoreTestCase):
"""
Test course enrollment data aggregation.
"""
USERNAME = "Bob"
EMAIL = "bob@example.com"
PASSWORD = "edx"
def setUp(self):
"""Create a course and user, then log in. """
super(EnrollmentDataTest, self).setUp()
self.course = CourseFactory.create()
self.user = UserFactory.create(username=self.USERNAME, email=self.EMAIL, password=self.PASSWORD)
self.client.login(username=self.USERNAME, password=self.PASSWORD)
@ddt.data(
# Default (no course modes in the database)
# Expect that users are automatically enrolled as "honor".
([], 'honor'),
# Audit / Verified / Honor
# We should always go to the "choose your course" page.
# We should also be enrolled as "honor" by default.
(['honor', 'verified', 'audit'], 'honor'),
)
@ddt.unpack
def test_enroll(self, course_modes, enrollment_mode):
# Create the course modes (if any) required for this test case
self._create_course_modes(course_modes)
enrollment = data.create_course_enrollment(
self.user.username,
six.text_type(self.course.id),
enrollment_mode,
True
)
self.assertTrue(CourseEnrollment.is_enrolled(self.user, self.course.id))
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(course_mode, enrollment_mode)
# Confirm the returned enrollment and the data match up.
self.assertEqual(course_mode, enrollment['mode'])
self.assertEqual(is_active, enrollment['is_active'])
self.assertEqual(self.course.display_name_with_default, enrollment['course_details']['course_name'])
def test_unenroll(self):
# Enroll the user in the course
CourseEnrollment.enroll(self.user, self.course.id, mode="honor")
enrollment = data.update_course_enrollment(
self.user.username,
six.text_type(self.course.id),
is_active=False
)
# Determine that the returned enrollment is inactive.
self.assertFalse(enrollment['is_active'])
# Expect that we're no longer enrolled
self.assertFalse(CourseEnrollment.is_enrolled(self.user, self.course.id))
@ddt.data(
# No course modes, no course enrollments.
([]),
# Audit / Verified / Honor course modes, with three course enrollments.
(['honor', 'verified', 'audit']),
)
def test_get_course_info(self, course_modes):
self._create_course_modes(course_modes, course=self.course)
result_course = data.get_course_enrollment_info(six.text_type(self.course.id))
result_slugs = [mode['slug'] for mode in result_course['course_modes']]
for course_mode in course_modes:
self.assertIn(course_mode, result_slugs)
@ddt.data(
# No course modes, no course enrollments.
([], []),
# Audit / Verified / Honor course modes, with three course enrollments.
(['honor', 'verified', 'audit'], ['1', '2', '3']),
)
@ddt.unpack
def test_get_course_enrollments(self, course_modes, course_numbers):
# Create all the courses
created_courses = []
for course_number in course_numbers:
created_courses.append(CourseFactory.create(number=course_number))
created_enrollments = []
for course in created_courses:
self._create_course_modes(course_modes, course=course)
# Create the original enrollment.
created_enrollments.append(data.create_course_enrollment(
self.user.username,
six.text_type(course.id),
'honor',
True
))
# Compare the created enrollments with the results
# from the get enrollments request.
results = data.get_course_enrollments(self.user.username)
self.assertEqual(results, created_enrollments)
# Now create a course enrollment with some invalid course (does
# not exist in database) for the user and check that the method
# 'get_course_enrollments' ignores course enrollments for invalid
# or deleted courses
CourseEnrollment.objects.create(
user=self.user,
course_id='InvalidOrg/InvalidCourse/InvalidRun',
mode='honor',
is_active=True
)
updated_results = data.get_course_enrollments(self.user.username)
self.assertEqual(results, updated_results)
def test_get_enrollments_including_inactive(self):
""" Verify that if 'include_inactive' is True, all enrollments
are returned including inactive.
"""
course_modes, course_numbers = ['honor', 'verified', 'audit'], ['1', '2', '3']
created_courses = []
for course_number in course_numbers:
created_courses.append(CourseFactory.create(number=course_number))
created_enrollments = []
for course in created_courses:
self._create_course_modes(course_modes, course=course)
# Create the original enrollment.
created_enrollments.append(data.create_course_enrollment(
self.user.username,
six.text_type(course.id),
'honor',
True
))
# deactivate one enrollment
data.update_course_enrollment(
self.user.username,
six.text_type(created_courses[0].id),
'honor',
False
)
# by default in-active enrollment will be excluded.
results = data.get_course_enrollments(self.user.username)
self.assertNotEqual(len(results), len(created_enrollments))
# we can get all enrollments including inactive by passing "include_inactive"
results = data.get_course_enrollments(self.user.username, include_inactive=True)
self.assertEqual(len(results), len(created_enrollments))
@ddt.data(
# Default (no course modes in the database)
# Expect that users are automatically enrolled as "honor".
([], 'honor'),
# Audit / Verified / Honor
# We should always go to the "choose your course" page.
# We should also be enrolled as "honor" by default.
(['honor', 'verified', 'audit'], 'verified'),
)
@ddt.unpack
def test_get_course_enrollment(self, course_modes, enrollment_mode):
self._create_course_modes(course_modes)
# Try to get an enrollment before it exists.
result = data.get_course_enrollment(self.user.username, six.text_type(self.course.id))
self.assertIsNone(result)
# Create the original enrollment.
enrollment = data.create_course_enrollment(
self.user.username,
six.text_type(self.course.id),
enrollment_mode,
True
)
# Get the enrollment and compare it to the original.
result = data.get_course_enrollment(self.user.username, six.text_type(self.course.id))
self.assertEqual(self.user.username, result['user'])
self.assertEqual(enrollment, result)
@ddt.data(
# Default (no course modes in the database)
# Expect that users are automatically enrolled as "honor".
([], 'honor'),
# Audit / Verified / Honor
# We should always go to the "choose your course" page.
# We should also be enrolled as "honor" by default.
(['honor', 'verified', 'audit'], 'verified'),
)
@ddt.unpack
def test_get_user_enrollments(self, course_modes, enrollment_mode):
self._create_course_modes(course_modes)
# Try to get enrollments before they exist.
result = data.get_user_enrollments(self.course.id)
self.assertFalse(result.exists())
# Create 10 test users to enroll in the course
users = []
for i in range(10):
users.append(UserFactory.create(
username=self.USERNAME + str(i),
email=self.EMAIL + str(i),
password=self.PASSWORD + str(i)
))
# Create the original enrollments.
created_enrollments = []
for user in users:
created_enrollments.append(data.create_course_enrollment(
user.username,
six.text_type(self.course.id),
enrollment_mode,
True
))
# Compare the created enrollments with the results
# from the get user enrollments request.
results = data.get_user_enrollments(
self.course.id
)
self.assertTrue(result.exists())
self.assertEqual(CourseEnrollmentSerializer(results, many=True).data, created_enrollments)
@ddt.data(
# Default (no course modes in the database)
# Expect that users are automatically enrolled as "honor".
([], 'credit'),
# Audit / Verified / Honor
# We should always go to the "choose your course" page.
# We should also be enrolled as "honor" by default.
(['honor', 'verified', 'audit', 'credit'], 'credit'),
)
@ddt.unpack
def test_add_or_update_enrollment_attr(self, course_modes, enrollment_mode):
# Create the course modes (if any) required for this test case
self._create_course_modes(course_modes)
data.create_course_enrollment(self.user.username, six.text_type(self.course.id), enrollment_mode, True)
enrollment_attributes = [
{
"namespace": "credit",
"name": "provider_id",
"value": "hogwarts",
}
]
data.add_or_update_enrollment_attr(self.user.username, six.text_type(self.course.id), enrollment_attributes)
enrollment_attr = data.get_enrollment_attributes(self.user.username, six.text_type(self.course.id))
self.assertEqual(enrollment_attr[0], enrollment_attributes[0])
enrollment_attributes = [
{
"namespace": "credit",
"name": "provider_id",
"value": "ASU",
}
]
data.add_or_update_enrollment_attr(self.user.username, six.text_type(self.course.id), enrollment_attributes)
enrollment_attr = data.get_enrollment_attributes(self.user.username, six.text_type(self.course.id))
self.assertEqual(enrollment_attr[0], enrollment_attributes[0])
def test_non_existent_course(self):
with pytest.raises(CourseNotFoundError):
data.get_course_enrollment_info("this/is/bananas")
def _create_course_modes(self, course_modes, course=None):
"""Create the course modes required for a test. """
course_id = course.id if course else self.course.id
for mode_slug in course_modes:
CourseModeFactory.create(
course_id=course_id,
mode_slug=mode_slug,
mode_display_name=mode_slug,
)
def test_enrollment_for_non_existent_user(self):
with pytest.raises(UserNotFoundError):
data.create_course_enrollment("some_fake_user", six.text_type(self.course.id), 'honor', True)
def test_enrollment_for_non_existent_course(self):
with pytest.raises(CourseNotFoundError):
data.create_course_enrollment(self.user.username, "some/fake/course", 'honor', True)
@patch.object(CourseEnrollment, "enroll")
def test_enrollment_for_closed_course(self, mock_enroll):
mock_enroll.side_effect = EnrollmentClosedError("Bad things happened")
with pytest.raises(CourseEnrollmentClosedError):
data.create_course_enrollment(self.user.username, six.text_type(self.course.id), 'honor', True)
@patch.object(CourseEnrollment, "enroll")
def test_enrollment_for_full_course(self, mock_enroll):
mock_enroll.side_effect = CourseFullError("Bad things happened")
with pytest.raises(CourseEnrollmentFullError):
data.create_course_enrollment(self.user.username, six.text_type(self.course.id), 'honor', True)
@patch.object(CourseEnrollment, "enroll")
def test_enrollment_for_enrolled_course(self, mock_enroll):
mock_enroll.side_effect = AlreadyEnrolledError("Bad things happened")
with pytest.raises(CourseEnrollmentExistsError):
data.create_course_enrollment(self.user.username, six.text_type(self.course.id), 'honor', True)
def test_update_for_non_existent_user(self):
with pytest.raises(UserNotFoundError):
data.update_course_enrollment("some_fake_user", six.text_type(self.course.id), is_active=False)
def test_update_for_non_existent_course(self):
enrollment = data.update_course_enrollment(self.user.username, "some/fake/course", is_active=False)
self.assertIsNone(enrollment)
def test_get_course_with_expired_mode_included(self):
"""Verify that method returns expired modes if include_expired
is true."""
modes = ['honor', 'verified', 'audit']
self._create_course_modes(modes, course=self.course)
self._update_verified_mode_as_expired(self.course.id)
self.assert_enrollment_modes(modes, True)
def test_get_course_without_expired_mode_included(self):
"""Verify that method does not returns expired modes if include_expired
is false."""
self._create_course_modes(['honor', 'verified', 'audit'], course=self.course)
self._update_verified_mode_as_expired(self.course.id)
self.assert_enrollment_modes(['audit', 'honor'], False)
def _update_verified_mode_as_expired(self, course_id):
"""Dry method to change verified mode expiration."""
mode = CourseMode.objects.get(course_id=course_id, mode_slug=CourseMode.VERIFIED)
mode.expiration_datetime = datetime.datetime(year=1970, month=1, day=1, tzinfo=UTC)
mode.save()
def assert_enrollment_modes(self, expected_modes, include_expired):
"""Get enrollment data and assert response with expected modes."""
result_course = data.get_course_enrollment_info(six.text_type(self.course.id), include_expired=include_expired)
result_slugs = [mode['slug'] for mode in result_course['course_modes']]
for course_mode in expected_modes:
self.assertIn(course_mode, result_slugs)
if not include_expired:
self.assertNotIn('verified', result_slugs)
def test_get_roles(self):
"""Create a role for a user, then get it"""
expected_role = CourseAccessRoleFactory.create(
course_id=self.course.id, user=self.user, role="SuperCoolTestRole",
)
roles = data.get_user_roles(self.user.username)
self.assertEqual(roles, {expected_role})
def test_get_roles_no_roles(self):
"""Get roles for a user who has no roles"""
roles = data.get_user_roles(self.user.username)
self.assertEqual(roles, set())
def test_get_roles_invalid_user(self):
"""Get roles for a user that doesn't exist"""
with pytest.raises(UserNotFoundError):
data.get_user_roles("i_dont_exist_and_should_raise_an_error")

View File

@@ -0,0 +1,1683 @@
# pylint: disable=missing-docstring,redefined-outer-name
"""
Tests for user enrollment.
"""
from __future__ import absolute_import
import datetime
import itertools
import json
import unittest
import ddt
import httpretty
import pytz
import six
from django.conf import settings
from django.core.cache import cache
from django.core.exceptions import ImproperlyConfigured
from django.core.handlers.wsgi import WSGIRequest
from django.test import Client
from django.test.utils import override_settings
from django.urls import reverse
from freezegun import freeze_time
from mock import patch
from rest_framework import status
from rest_framework.test import APITestCase
from six import text_type
from six.moves import range
from course_modes.models import CourseMode
from course_modes.tests.factories import CourseModeFactory
from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
from openedx.core.djangoapps.course_groups import cohorts
from openedx.core.djangoapps.embargo.models import Country, CountryAccessRule, RestrictedCourse
from openedx.core.djangoapps.embargo.test_utils import restrict_course
from openedx.core.djangoapps.enrollments import api, data
from openedx.core.djangoapps.enrollments.errors import CourseEnrollmentError
from openedx.core.djangoapps.enrollments.views import EnrollmentUserThrottle
from openedx.core.djangoapps.oauth_dispatch.jwt import create_jwt_for_user
from openedx.core.djangoapps.user_api.models import RetirementState, UserOrgTag, UserRetirementStatus
from openedx.core.lib.django_test_client_utils import get_absolute_url
from openedx.features.enterprise_support.tests import FAKE_ENTERPRISE_CUSTOMER
from openedx.features.enterprise_support.tests.mixins.enterprise import EnterpriseServiceMockMixin
from student.models import CourseEnrollment
from student.roles import CourseStaffRole
from student.tests.factories import AdminFactory, SuperuserFactory, UserFactory
from util.models import RateLimitConfiguration
from util.testing import UrlResetMixin
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory, check_mongo_calls_range
class EnrollmentTestMixin(object):
""" Mixin with methods useful for testing enrollments. """
API_KEY = "i am a key"
def assert_enrollment_status(
self,
course_id=None,
username=None,
expected_status=status.HTTP_200_OK,
email_opt_in=None,
as_server=False,
mode=CourseMode.DEFAULT_MODE_SLUG,
is_active=None,
enrollment_attributes=None,
min_mongo_calls=0,
max_mongo_calls=0,
linked_enterprise_customer=None,
cohort=None,
):
"""
Enroll in the course and verify the response's status code. If the expected status is 200, also validates
the response content.
Returns
Response
"""
course_id = course_id or six.text_type(self.course.id)
username = username or self.user.username
data = {
'mode': mode,
'course_details': {
'course_id': course_id
},
'user': username,
'enrollment_attributes': enrollment_attributes
}
if is_active is not None:
data['is_active'] = is_active
if email_opt_in is not None:
data['email_opt_in'] = email_opt_in
if linked_enterprise_customer is not None:
data['linked_enterprise_customer'] = linked_enterprise_customer
if cohort is not None:
data['cohort'] = cohort
extra = {}
if as_server:
extra['HTTP_X_EDX_API_KEY'] = self.API_KEY
# Verify that the modulestore is queried as expected.
with check_mongo_calls_range(min_finds=min_mongo_calls, max_finds=max_mongo_calls):
with patch('openedx.core.djangoapps.enrollments.views.audit_log') as mock_audit_log:
url = reverse('courseenrollments')
response = self.client.post(url, json.dumps(data), content_type='application/json', **extra)
self.assertEqual(response.status_code, expected_status)
if expected_status == status.HTTP_200_OK:
data = json.loads(response.content)
self.assertEqual(course_id, data['course_details']['course_id'])
if mode is not None:
self.assertEqual(mode, data['mode'])
if is_active is not None:
self.assertEqual(is_active, data['is_active'])
else:
self.assertTrue(data['is_active'])
if as_server:
# Verify that an audit message was logged.
self.assertTrue(mock_audit_log.called)
# If multiple enrollment calls are made in the scope of a
# single test, we want to validate that audit messages are
# logged for each call.
mock_audit_log.reset_mock()
return response
def assert_enrollment_activation(self, expected_activation, expected_mode):
"""Change an enrollment's activation and verify its activation and mode are as expected."""
self.assert_enrollment_status(
as_server=True,
mode=expected_mode,
is_active=expected_activation,
expected_status=status.HTTP_200_OK
)
actual_mode, actual_activation = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertEqual(actual_activation, expected_activation)
self.assertEqual(actual_mode, expected_mode)
def _get_enrollments(self):
"""Retrieve the enrollment list for the current user. """
resp = self.client.get(reverse("courseenrollments"))
return json.loads(resp.content)
@override_settings(EDX_API_KEY="i am a key")
@ddt.ddt
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
class EnrollmentTest(EnrollmentTestMixin, ModuleStoreTestCase, APITestCase, EnterpriseServiceMockMixin):
"""
Test user enrollment, especially with different course modes.
"""
USERNAME = "Bob"
EMAIL = "bob@example.com"
PASSWORD = "edx"
OTHER_USERNAME = "Jane"
OTHER_EMAIL = "jane@example.com"
ENABLED_CACHES = ['default', 'mongo_metadata_inheritance', 'loc_cache']
ENABLED_SIGNALS = ['course_published']
def setUp(self):
""" Create a course and user, then log in. """
super(EnrollmentTest, self).setUp()
self.rate_limit_config = RateLimitConfiguration.current()
self.rate_limit_config.enabled = False
self.rate_limit_config.save()
throttle = EnrollmentUserThrottle()
self.rate_limit, __ = throttle.parse_rate(throttle.rate)
# Pass emit_signals when creating the course so it would be cached
# as a CourseOverview. Enrollments require a cached CourseOverview.
self.course = CourseFactory.create(emit_signals=True)
self.user = UserFactory.create(
username=self.USERNAME,
email=self.EMAIL,
password=self.PASSWORD,
)
self.other_user = UserFactory.create(
username=self.OTHER_USERNAME,
email=self.OTHER_EMAIL,
password=self.PASSWORD,
)
self.client.login(username=self.USERNAME, password=self.PASSWORD)
@ddt.data(
# Default (no course modes in the database)
# Expect that users are automatically enrolled as the default
([], CourseMode.DEFAULT_MODE_SLUG),
# Audit / Verified
# We should always go to the "choose your course" page.
# We should also be enrolled as the default.
([CourseMode.VERIFIED, CourseMode.AUDIT], CourseMode.DEFAULT_MODE_SLUG),
)
@ddt.unpack
def test_enroll(self, course_modes, enrollment_mode):
# Create the course modes (if any) required for this test case
for mode_slug in course_modes:
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=mode_slug,
mode_display_name=mode_slug,
)
# Create an enrollment
resp = self.assert_enrollment_status()
# Verify that the response contains the correct course_name
data = json.loads(resp.content)
self.assertEqual(self.course.display_name_with_default, data['course_details']['course_name'])
# Verify that the enrollment was created correctly
self.assertTrue(CourseEnrollment.is_enrolled(self.user, self.course.id))
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(course_mode, enrollment_mode)
def test_check_enrollment(self):
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=CourseMode.DEFAULT_MODE_SLUG,
mode_display_name=CourseMode.DEFAULT_MODE_SLUG,
)
# Create an enrollment
self.assert_enrollment_status()
resp = self.client.get(
reverse(
'courseenrollment',
kwargs={'username': self.user.username, "course_id": six.text_type(self.course.id)},
)
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = json.loads(resp.content)
self.assertEqual(six.text_type(self.course.id), data['course_details']['course_id'])
self.assertEqual(self.course.display_name_with_default, data['course_details']['course_name'])
self.assertEqual(CourseMode.DEFAULT_MODE_SLUG, data['mode'])
self.assertTrue(data['is_active'])
@ddt.data(
(True, u"True"),
(False, u"False"),
(None, None)
)
@ddt.unpack
def test_email_opt_in_true(self, opt_in, pref_value):
"""
Verify that the email_opt_in parameter sets the underlying flag.
And that if the argument is not present, then it does not affect the flag
"""
def _assert_no_opt_in_set():
""" Check the tag doesn't exit"""
with self.assertRaises(UserOrgTag.DoesNotExist):
UserOrgTag.objects.get(user=self.user, org=self.course.id.org, key="email-optin")
_assert_no_opt_in_set()
self.assert_enrollment_status(email_opt_in=opt_in)
if opt_in is None:
_assert_no_opt_in_set()
else:
preference = UserOrgTag.objects.get(user=self.user, org=self.course.id.org, key="email-optin")
self.assertEquals(preference.value, pref_value)
def test_enroll_prof_ed(self):
# Create the prod ed mode.
CourseModeFactory.create(
course_id=self.course.id,
mode_slug='professional',
mode_display_name='Professional Education',
)
# Enroll in the course, this will fail if the mode is not explicitly professional.
resp = self.assert_enrollment_status(expected_status=status.HTTP_400_BAD_REQUEST)
# While the enrollment wrong is invalid, the response content should have
# all the valid enrollment modes.
data = json.loads(resp.content)
self.assertEqual(six.text_type(self.course.id), data['course_details']['course_id'])
self.assertEqual(1, len(data['course_details']['course_modes']))
self.assertEqual('professional', data['course_details']['course_modes'][0]['slug'])
def test_user_not_specified(self):
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=CourseMode.DEFAULT_MODE_SLUG,
mode_display_name=CourseMode.DEFAULT_MODE_SLUG,
)
# Create an enrollment
self.assert_enrollment_status()
resp = self.client.get(
reverse('courseenrollment', kwargs={"course_id": six.text_type(self.course.id)})
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = json.loads(resp.content)
self.assertEqual(six.text_type(self.course.id), data['course_details']['course_id'])
self.assertEqual(CourseMode.DEFAULT_MODE_SLUG, data['mode'])
self.assertTrue(data['is_active'])
def test_user_not_authenticated(self):
# Log out, so we're no longer authenticated
self.client.logout()
# Try to enroll, this should fail.
self.assert_enrollment_status(expected_status=status.HTTP_401_UNAUTHORIZED)
def test_user_not_activated(self):
# Log out the default user, Bob.
self.client.logout()
# Create a user account
self.user = UserFactory.create(
username="inactive",
email="inactive@example.com",
password=self.PASSWORD,
is_active=True
)
# Log in with the unactivated account
self.client.login(username="inactive", password=self.PASSWORD)
# Deactivate the user. Has to be done after login to get the user into the
# request and properly logged in.
self.user.is_active = False
self.user.save()
# Enrollment should succeed, even though we haven't authenticated.
self.assert_enrollment_status()
def test_user_does_not_match_url(self):
# Try to enroll a user that is not the authenticated user.
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=CourseMode.DEFAULT_MODE_SLUG,
mode_display_name=CourseMode.DEFAULT_MODE_SLUG,
)
self.assert_enrollment_status(username=self.other_user.username, expected_status=status.HTTP_404_NOT_FOUND)
# Verify that the server still has access to this endpoint.
self.client.logout()
self.assert_enrollment_status(username=self.other_user.username, as_server=True)
def _assert_enrollments_visible_in_list(self, courses, use_server_key=False):
"""
Check that the list of enrollments of self.user returned for the currently logged in user
matches the list of courses passed in in 'courses'.
"""
kwargs = {}
if use_server_key:
kwargs.update(HTTP_X_EDX_API_KEY=self.API_KEY)
response = self.client.get(reverse('courseenrollments'), {'user': self.user.username}, **kwargs)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = json.loads(response.content)
self.assertItemsEqual(
[(datum['course_details']['course_id'], datum['course_details']['course_name']) for datum in data],
[(six.text_type(course.id), course.display_name_with_default) for course in courses]
)
def test_enrollment_list_permissions(self):
"""
Test that the correct list of enrollments is returned, depending on the permissions of the
requesting user.
"""
# Create another course, and enroll self.user in both courses.
other_course = CourseFactory.create(emit_signals=True)
for course in self.course, other_course:
CourseModeFactory.create(
course_id=six.text_type(course.id),
mode_slug=CourseMode.DEFAULT_MODE_SLUG,
mode_display_name=CourseMode.DEFAULT_MODE_SLUG,
)
self.assert_enrollment_status(
course_id=six.text_type(course.id),
max_mongo_calls=0,
)
# Verify the user himself can see both of his enrollments.
self._assert_enrollments_visible_in_list([self.course, other_course])
# Verify that self.other_user can't see any of the enrollments.
self.client.login(username=self.OTHER_USERNAME, password=self.PASSWORD)
self._assert_enrollments_visible_in_list([])
# Create a staff user for self.course (but nor for other_course) and log her in.
staff_user = UserFactory.create(username='staff', email='staff@example.com', password=self.PASSWORD)
CourseStaffRole(self.course.id).add_users(staff_user)
self.client.login(username='staff', password=self.PASSWORD)
# Verify that she can see only the enrollment in the course she has staff privileges for.
self._assert_enrollments_visible_in_list([self.course])
# Create a global staff user, and verify she can see all enrollments.
AdminFactory(username='global_staff', email='global_staff@example.com', password=self.PASSWORD)
self.client.login(username='global_staff', password=self.PASSWORD)
self._assert_enrollments_visible_in_list([self.course, other_course])
# Verify the server can see all enrollments.
self.client.logout()
self._assert_enrollments_visible_in_list([self.course, other_course], use_server_key=True)
def test_user_does_not_match_param(self):
"""
The view should return status 404 if the enrollment username does not match the username of the user
making the request, unless the request is made by a staff user or with a server API key.
"""
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=CourseMode.HONOR,
mode_display_name=CourseMode.HONOR,
)
url = reverse('courseenrollment',
kwargs={'username': self.other_user.username, "course_id": six.text_type(self.course.id)})
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
# Verify that the server still has access to this endpoint.
self.client.logout()
response = self.client.get(url, **{'HTTP_X_EDX_API_KEY': self.API_KEY})
self.assertEqual(response.status_code, status.HTTP_200_OK)
# Verify staff have access to this endpoint
staff_user = UserFactory.create(password=self.PASSWORD, is_staff=True)
self.client.login(username=staff_user.username, password=self.PASSWORD)
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_get_course_details(self):
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=CourseMode.HONOR,
mode_display_name=CourseMode.HONOR,
sku='123',
bulk_sku="BULK123"
)
resp = self.client.get(
reverse('courseenrollmentdetails', kwargs={"course_id": six.text_type(self.course.id)})
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = json.loads(resp.content)
self.assertEqual(six.text_type(self.course.id), data['course_id'])
self.assertEqual(self.course.display_name_with_default, data['course_name'])
mode = data['course_modes'][0]
self.assertEqual(mode['slug'], CourseMode.HONOR)
self.assertEqual(mode['sku'], '123')
self.assertEqual(mode['bulk_sku'], 'BULK123')
self.assertEqual(mode['name'], CourseMode.HONOR)
def test_get_course_details_with_credit_course(self):
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=CourseMode.CREDIT_MODE,
mode_display_name=CourseMode.CREDIT_MODE,
)
resp = self.client.get(
reverse('courseenrollmentdetails', kwargs={"course_id": six.text_type(self.course.id)})
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = json.loads(resp.content)
self.assertEqual(six.text_type(self.course.id), data['course_id'])
mode = data['course_modes'][0]
self.assertEqual(mode['slug'], CourseMode.CREDIT_MODE)
self.assertEqual(mode['name'], CourseMode.CREDIT_MODE)
@ddt.data(
# NOTE: Studio requires a start date, but this is not
# enforced at the data layer, so we need to handle the case
# in which no dates are specified.
(None, None, None, None),
(datetime.datetime(2015, 1, 2, 3, 4, 5, tzinfo=pytz.UTC), None, "2015-01-02T03:04:05Z", None),
(None, datetime.datetime(2015, 1, 2, 3, 4, 5, tzinfo=pytz.UTC), None, "2015-01-02T03:04:05Z"),
(
datetime.datetime(2014, 6, 7, 8, 9, 10, tzinfo=pytz.UTC),
datetime.datetime(2015, 1, 2, 3, 4, 5, tzinfo=pytz.UTC),
"2014-06-07T08:09:10Z",
"2015-01-02T03:04:05Z",
),
)
@ddt.unpack
def test_get_course_details_course_dates(self, start_datetime, end_datetime, expected_start, expected_end):
course = CourseFactory.create(start=start_datetime, end=end_datetime)
# Load a CourseOverview. This initial load should result in a cache
# miss; the modulestore is queried and course metadata is cached.
__ = CourseOverview.get_from_id(course.id)
self.assert_enrollment_status(course_id=six.text_type(course.id))
# Check course details
url = reverse('courseenrollmentdetails', kwargs={"course_id": six.text_type(course.id)})
resp = self.client.get(url)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = json.loads(resp.content)
self.assertEqual(data['course_start'], expected_start)
self.assertEqual(data['course_end'], expected_end)
# Check enrollment course details
url = reverse('courseenrollment', kwargs={"course_id": six.text_type(course.id)})
resp = self.client.get(url)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = json.loads(resp.content)
self.assertEqual(data['course_details']['course_start'], expected_start)
self.assertEqual(data['course_details']['course_end'], expected_end)
# Check enrollment list course details
resp = self.client.get(reverse('courseenrollments'))
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = json.loads(resp.content)
self.assertEqual(data[0]['course_details']['course_start'], expected_start)
self.assertEqual(data[0]['course_details']['course_end'], expected_end)
def test_with_invalid_course_id(self):
self.assert_enrollment_status(
course_id='entirely/fake/course',
expected_status=status.HTTP_400_BAD_REQUEST,
min_mongo_calls=3,
max_mongo_calls=4
)
def test_get_enrollment_details_bad_course(self):
resp = self.client.get(
reverse('courseenrollmentdetails', kwargs={"course_id": "some/fake/course"})
)
self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
@patch.object(api, "get_enrollment")
def test_get_enrollment_internal_error(self, mock_get_enrollment):
mock_get_enrollment.side_effect = CourseEnrollmentError("Something bad happened.")
resp = self.client.get(
reverse(
'courseenrollment',
kwargs={'username': self.user.username, "course_id": six.text_type(self.course.id)},
)
)
self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
def test_enrollment_already_enrolled(self):
response = self.assert_enrollment_status()
repeat_response = self.assert_enrollment_status(expected_status=status.HTTP_200_OK)
self.assertEqual(json.loads(response.content), json.loads(repeat_response.content))
def test_get_enrollment_with_invalid_key(self):
resp = self.client.post(
reverse('courseenrollments'),
{
'course_details': {
'course_id': 'invalidcourse'
},
'user': self.user.username
},
format='json'
)
self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("No course ", resp.content)
def test_enrollment_throttle_for_user(self):
"""Make sure a user requests do not exceed the maximum number of requests"""
self.rate_limit_config.enabled = True
self.rate_limit_config.save()
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=CourseMode.DEFAULT_MODE_SLUG,
mode_display_name=CourseMode.DEFAULT_MODE_SLUG,
)
for attempt in range(self.rate_limit + 2):
expected_status = status.HTTP_429_TOO_MANY_REQUESTS if attempt >= self.rate_limit else status.HTTP_200_OK
self.assert_enrollment_status(expected_status=expected_status)
@ddt.data('staff', 'user')
def test_enrollment_throttle_is_set_correctly(self, user_scope):
""" Make sure throttle rate is set correctly for different user scopes. """
self.rate_limit_config.enabled = True
self.rate_limit_config.save()
throttle = EnrollmentUserThrottle()
throttle.scope = user_scope
try:
throttle.parse_rate(throttle.get_rate())
except ImproperlyConfigured:
self.fail(u"No throttle rate set for {}".format(user_scope))
def test_create_enrollment_with_cohort(self):
"""Enroll in the course, and also add to a cohort."""
# Create a cohort
cohort_name = 'masters'
cohorts.set_course_cohorted(self.course.id, True)
cohorts.add_cohort(self.course.id, cohort_name, 'test')
# Create an enrollment
self.assert_enrollment_status(cohort=cohort_name)
self.assertTrue(CourseEnrollment.is_enrolled(self.user, self.course.id))
_, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(cohorts.get_cohort(self.user, self.course.id, assign=False).name, cohort_name)
def test_create_enrollment_with_wrong_cohort(self):
"""Enroll in the course, and also add to a cohort."""
# Create a cohort
cohorts.set_course_cohorted(self.course.id, True)
cohorts.add_cohort(self.course.id, 'masters', 'test')
# Create an enrollment
self.assert_enrollment_status(cohort='missing', expected_status=status.HTTP_400_BAD_REQUEST)
def test_create_enrollment_with_mode(self):
"""With the right API key, create a new enrollment with a mode set other than the default."""
# Create a professional ed course mode.
CourseModeFactory.create(
course_id=self.course.id,
mode_slug='professional',
mode_display_name='professional',
)
# Create an enrollment
self.assert_enrollment_status(as_server=True, mode='professional')
self.assertTrue(CourseEnrollment.is_enrolled(self.user, self.course.id))
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(course_mode, 'professional')
def test_enrollment_includes_expired_verified(self):
"""With the right API key, request that expired course verifications are still returned. """
# Create a honor mode for a course.
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=CourseMode.HONOR,
mode_display_name=CourseMode.HONOR,
)
# Create a verified mode for a course.
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=CourseMode.VERIFIED,
mode_display_name=CourseMode.VERIFIED,
expiration_datetime='1970-01-01 05:00:00Z'
)
# Passes the include_expired parameter to the API call
v_response = self.client.get(
reverse(
'courseenrollmentdetails',
kwargs={"course_id": six.text_type(self.course.id)}
),
{'include_expired': True},
)
v_data = json.loads(v_response.content)
# Ensure that both course modes are returned
self.assertEqual(len(v_data['course_modes']), 2)
# Omits the include_expired parameter from the API call
h_response = self.client.get(
reverse('courseenrollmentdetails', kwargs={"course_id": six.text_type(self.course.id)}),
)
h_data = json.loads(h_response.content)
# Ensure that only one course mode is returned and that it is honor
self.assertEqual(len(h_data['course_modes']), 1)
self.assertEqual(h_data['course_modes'][0]['slug'], CourseMode.HONOR)
def test_update_enrollment_with_mode(self):
"""With the right API key, update an existing enrollment with a new mode. """
# Create an honor and verified mode for a course. This allows an update.
for mode in [CourseMode.DEFAULT_MODE_SLUG, CourseMode.VERIFIED]:
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=mode,
mode_display_name=mode,
)
# Create an enrollment
self.assert_enrollment_status(as_server=True)
# Check that the enrollment is default.
self.assertTrue(CourseEnrollment.is_enrolled(self.user, self.course.id))
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(course_mode, CourseMode.DEFAULT_MODE_SLUG)
# Check that the enrollment upgraded to verified.
self.assert_enrollment_status(as_server=True, mode=CourseMode.VERIFIED, expected_status=status.HTTP_200_OK)
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(course_mode, CourseMode.VERIFIED)
def test_enrollment_with_credit_mode(self):
"""With the right API key, update an existing enrollment with credit
mode and set enrollment attributes.
"""
for mode in [CourseMode.DEFAULT_MODE_SLUG, CourseMode.CREDIT_MODE]:
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=mode,
mode_display_name=mode,
)
# Create an enrollment
self.assert_enrollment_status(as_server=True)
# Check that the enrollment is the default.
self.assertTrue(CourseEnrollment.is_enrolled(self.user, self.course.id))
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(course_mode, CourseMode.DEFAULT_MODE_SLUG)
# Check that the enrollment upgraded to credit.
enrollment_attributes = [{
"namespace": "credit",
"name": "provider_id",
"value": "hogwarts",
}]
self.assert_enrollment_status(
as_server=True,
mode=CourseMode.CREDIT_MODE,
expected_status=status.HTTP_200_OK,
enrollment_attributes=enrollment_attributes
)
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(course_mode, CourseMode.CREDIT_MODE)
def test_enrollment_with_invalid_attr(self):
"""Check response status is bad request when invalid enrollment
attributes are passed
"""
for mode in [CourseMode.DEFAULT_MODE_SLUG, CourseMode.CREDIT_MODE]:
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=mode,
mode_display_name=mode,
)
# Create an enrollment
self.assert_enrollment_status(as_server=True)
# Check that the enrollment is the default.
self.assertTrue(CourseEnrollment.is_enrolled(self.user, self.course.id))
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(course_mode, CourseMode.DEFAULT_MODE_SLUG)
# Check that the enrollment upgraded to credit.
enrollment_attributes = [{
"namespace": "credit",
"name": "invalid",
"value": "hogwarts",
}]
self.assert_enrollment_status(
as_server=True,
mode=CourseMode.CREDIT_MODE,
expected_status=status.HTTP_400_BAD_REQUEST,
enrollment_attributes=enrollment_attributes
)
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(course_mode, CourseMode.DEFAULT_MODE_SLUG)
def test_downgrade_enrollment_with_mode(self):
"""With the right API key, downgrade an existing enrollment with a new mode. """
# Create an honor and verified mode for a course. This allows an update.
for mode in [CourseMode.DEFAULT_MODE_SLUG, CourseMode.VERIFIED]:
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=mode,
mode_display_name=mode,
)
# Create a 'verified' enrollment
self.assert_enrollment_status(as_server=True, mode=CourseMode.VERIFIED)
# Check that the enrollment is verified.
self.assertTrue(CourseEnrollment.is_enrolled(self.user, self.course.id))
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(course_mode, CourseMode.VERIFIED)
# Check that the enrollment was downgraded to the default mode.
self.assert_enrollment_status(
as_server=True,
mode=CourseMode.DEFAULT_MODE_SLUG,
expected_status=status.HTTP_200_OK
)
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(course_mode, CourseMode.DEFAULT_MODE_SLUG)
@ddt.data(
((CourseMode.DEFAULT_MODE_SLUG, ), CourseMode.DEFAULT_MODE_SLUG),
((CourseMode.DEFAULT_MODE_SLUG, CourseMode.VERIFIED), CourseMode.DEFAULT_MODE_SLUG),
((CourseMode.DEFAULT_MODE_SLUG, CourseMode.VERIFIED), CourseMode.VERIFIED),
((CourseMode.PROFESSIONAL, ), CourseMode.PROFESSIONAL),
((CourseMode.NO_ID_PROFESSIONAL_MODE, ), CourseMode.NO_ID_PROFESSIONAL_MODE),
((CourseMode.VERIFIED, CourseMode.CREDIT_MODE), CourseMode.VERIFIED),
((CourseMode.VERIFIED, CourseMode.CREDIT_MODE), CourseMode.CREDIT_MODE),
)
@ddt.unpack
def test_deactivate_enrollment(self, configured_modes, selected_mode):
"""With the right API key, deactivate (i.e., unenroll from) an existing enrollment."""
# Configure a set of modes for the course.
for mode in configured_modes:
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=mode,
mode_display_name=mode,
)
# Create an enrollment with the selected mode.
self.assert_enrollment_status(as_server=True, mode=selected_mode)
# Check that the enrollment has the correct mode and is active.
self.assertTrue(CourseEnrollment.is_enrolled(self.user, self.course.id))
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(course_mode, selected_mode)
# Verify that a non-Boolean enrollment status is treated as invalid.
self.assert_enrollment_status(
as_server=True,
mode=None,
is_active='foo',
expected_status=status.HTTP_400_BAD_REQUEST
)
# Verify that the enrollment has been deactivated, and that the mode is unchanged.
self.assert_enrollment_activation(False, selected_mode)
# Verify that enrollment deactivation is idempotent.
self.assert_enrollment_activation(False, selected_mode)
# Verify that omitting the mode returns 400 for course configurations
# in which the default mode doesn't exist.
expected_status = (
status.HTTP_200_OK
if CourseMode.DEFAULT_MODE_SLUG in configured_modes
else status.HTTP_400_BAD_REQUEST
)
self.assert_enrollment_status(
as_server=True,
is_active=False,
expected_status=expected_status,
)
def test_deactivate_enrollment_expired_mode(self):
"""Verify that an enrollment in an expired mode can be deactivated."""
for mode in (CourseMode.HONOR, CourseMode.VERIFIED):
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=mode,
mode_display_name=mode,
)
# Create verified enrollment.
self.assert_enrollment_status(as_server=True, mode=CourseMode.VERIFIED)
# Change verified mode expiration.
mode = CourseMode.objects.get(course_id=self.course.id, mode_slug=CourseMode.VERIFIED)
mode.expiration_datetime = datetime.datetime(year=1970, month=1, day=1, tzinfo=pytz.utc)
mode.save()
# Deactivate enrollment.
self.assert_enrollment_activation(False, CourseMode.VERIFIED)
def test_change_mode_from_user(self):
"""Users should not be able to alter the enrollment mode on an enrollment. """
# Create a default and a verified mode for a course. This allows an update.
for mode in [CourseMode.DEFAULT_MODE_SLUG, CourseMode.VERIFIED]:
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=mode,
mode_display_name=mode,
)
# Create an enrollment
self.assert_enrollment_status()
# Check that the enrollment is honor.
self.assertTrue(CourseEnrollment.is_enrolled(self.user, self.course.id))
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(course_mode, CourseMode.DEFAULT_MODE_SLUG)
# Get a 403 response when trying to upgrade yourself.
self.assert_enrollment_status(mode=CourseMode.VERIFIED, expected_status=status.HTTP_403_FORBIDDEN)
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(course_mode, CourseMode.DEFAULT_MODE_SLUG)
@ddt.data(*itertools.product(
(CourseMode.HONOR, CourseMode.VERIFIED),
(CourseMode.HONOR, CourseMode.VERIFIED),
(True, False),
(True, False),
))
@ddt.unpack
def test_change_mode_from_server(self, old_mode, new_mode, old_is_active, new_is_active):
"""
Server-to-server calls should be allowed to change the mode of any
enrollment, as long as the enrollment is not being deactivated during
the same call (this is assumed to be an error on the client's side).
"""
for mode in [CourseMode.HONOR, CourseMode.VERIFIED]:
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=mode,
mode_display_name=mode,
)
# Set up the initial enrollment
self.assert_enrollment_status(as_server=True, mode=old_mode, is_active=old_is_active)
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertEqual(is_active, old_is_active)
self.assertEqual(course_mode, old_mode)
expected_status = status.HTTP_400_BAD_REQUEST if (
old_mode != new_mode and
old_is_active != new_is_active and
not new_is_active
) else status.HTTP_200_OK
# simulate the server-server api call under test
response = self.assert_enrollment_status(
as_server=True,
mode=new_mode,
is_active=new_is_active,
expected_status=expected_status,
)
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
if expected_status == status.HTTP_400_BAD_REQUEST:
# nothing should have changed
self.assertEqual(is_active, old_is_active)
self.assertEqual(course_mode, old_mode)
# error message should contain specific text. Otto checks for this text in the message.
self.assertRegexpMatches(json.loads(response.content)['message'], 'Enrollment mode mismatch')
else:
# call should have succeeded
self.assertEqual(is_active, new_is_active)
self.assertEqual(course_mode, new_mode)
def test_change_mode_invalid_user(self):
"""
Attempts to change an enrollment for a non-existent user should result in an HTTP 404 for non-server users,
and HTTP 406 for server users.
"""
self.assert_enrollment_status(username='fake-user', expected_status=status.HTTP_404_NOT_FOUND, as_server=False)
self.assert_enrollment_status(username='fake-user', expected_status=status.HTTP_406_NOT_ACCEPTABLE,
as_server=True)
@ddt.data(
(True, CourseMode.VERIFIED),
(False, CourseMode.DEFAULT_MODE_SLUG)
)
@ddt.unpack
def test_update_enrollment_with_expired_mode(self, using_api_key, updated_mode):
"""Verify that if verified mode is expired than it's enrollment cannot be updated. """
for mode in [CourseMode.DEFAULT_MODE_SLUG, CourseMode.VERIFIED]:
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=mode,
mode_display_name=mode,
)
# Create an enrollment
self.assert_enrollment_status(as_server=True)
# Check that the enrollment is the default.
self.assertTrue(CourseEnrollment.is_enrolled(self.user, self.course.id))
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(course_mode, CourseMode.DEFAULT_MODE_SLUG)
# Change verified mode expiration.
mode = CourseMode.objects.get(course_id=self.course.id, mode_slug=CourseMode.VERIFIED)
mode.expiration_datetime = datetime.datetime(year=1970, month=1, day=1, tzinfo=pytz.utc)
mode.save()
self.assert_enrollment_status(
as_server=using_api_key,
mode=CourseMode.VERIFIED,
expected_status=status.HTTP_200_OK if using_api_key else status.HTTP_403_FORBIDDEN
)
course_mode, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, self.course.id)
self.assertTrue(is_active)
self.assertEqual(course_mode, updated_mode)
@httpretty.activate
@override_settings(ENTERPRISE_SERVICE_WORKER_USERNAME='enterprise_worker',
FEATURES=dict(ENABLE_ENTERPRISE_INTEGRATION=True))
@patch('openedx.features.enterprise_support.api.enterprise_customer_from_api')
def test_enterprise_course_enrollment_with_ec_uuid(self, mock_enterprise_customer_from_api):
"""Verify that the enrollment completes when the EnterpriseCourseEnrollment creation succeeds. """
UserFactory.create(
username='enterprise_worker',
email=self.EMAIL,
password=self.PASSWORD,
)
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=CourseMode.DEFAULT_MODE_SLUG,
mode_display_name=CourseMode.DEFAULT_MODE_SLUG,
)
consent_kwargs = {
'username': self.user.username,
'course_id': six.text_type(self.course.id),
'ec_uuid': 'this-is-a-real-uuid'
}
mock_enterprise_customer_from_api.return_value = FAKE_ENTERPRISE_CUSTOMER
self.mock_enterprise_course_enrollment_post_api()
self.mock_consent_missing(**consent_kwargs)
self.mock_consent_post(**consent_kwargs)
self.assert_enrollment_status(
expected_status=status.HTTP_200_OK,
as_server=True,
username='enterprise_worker',
linked_enterprise_customer='this-is-a-real-uuid',
)
self.assertEqual(
httpretty.last_request().path,
'/consent/api/v1/data_sharing_consent',
)
self.assertEqual(
httpretty.last_request().method,
httpretty.POST
)
def test_enrollment_attributes_always_written(self):
""" Enrollment attributes should always be written, regardless of whether
the enrollment is being created or updated.
"""
course_key = self.course.id
for mode in [CourseMode.DEFAULT_MODE_SLUG, CourseMode.VERIFIED]:
CourseModeFactory.create(
course_id=course_key,
mode_slug=mode,
mode_display_name=mode,
)
# Creating a new enrollment should write attributes
order_number = 'EDX-1000'
enrollment_attributes = [{
'namespace': 'order',
'name': 'order_number',
'value': order_number,
}]
mode = CourseMode.VERIFIED
self.assert_enrollment_status(
as_server=True,
is_active=True,
mode=mode,
enrollment_attributes=enrollment_attributes
)
enrollment = CourseEnrollment.objects.get(user=self.user, course_id=course_key)
self.assertTrue(enrollment.is_active)
self.assertEqual(enrollment.mode, CourseMode.VERIFIED)
self.assertEqual(enrollment.attributes.get(namespace='order', name='order_number').value, order_number)
# Updating an enrollment should update attributes
order_number = 'EDX-2000'
enrollment_attributes = [{
'namespace': 'order',
'name': 'order_number',
'value': order_number,
}]
mode = CourseMode.DEFAULT_MODE_SLUG
self.assert_enrollment_status(
as_server=True,
mode=mode,
enrollment_attributes=enrollment_attributes
)
enrollment.refresh_from_db()
self.assertTrue(enrollment.is_active)
self.assertEqual(enrollment.mode, mode)
self.assertEqual(enrollment.attributes.get(namespace='order', name='order_number').value, order_number)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
class EnrollmentEmbargoTest(EnrollmentTestMixin, UrlResetMixin, ModuleStoreTestCase):
"""Test that enrollment is blocked from embargoed countries. """
USERNAME = "Bob"
EMAIL = "bob@example.com"
PASSWORD = "edx"
URLCONF_MODULES = ['openedx.core.djangoapps.embargo']
@patch.dict(settings.FEATURES, {'EMBARGO': True})
def setUp(self):
""" Create a course and user, then log in. """
super(EnrollmentEmbargoTest, self).setUp()
self.course = CourseFactory.create()
# Load a CourseOverview. This initial load should result in a cache
# miss; the modulestore is queried and course metadata is cached.
__ = CourseOverview.get_from_id(self.course.id)
self.user = UserFactory.create(username=self.USERNAME, email=self.EMAIL, password=self.PASSWORD)
self.client.login(username=self.USERNAME, password=self.PASSWORD)
self.url = reverse('courseenrollments')
def _generate_data(self):
return json.dumps({
'course_details': {
'course_id': six.text_type(self.course.id)
},
'user': self.user.username
})
def assert_access_denied(self, user_message_path):
"""
Verify that the view returns HTTP status 403 and includes a URL in the response, and no enrollment is created.
"""
data = self._generate_data()
response = self.client.post(self.url, data, content_type='application/json')
# Expect an error response
self.assertEqual(response.status_code, 403)
# Expect that the redirect URL is included in the response
resp_data = json.loads(response.content)
user_message_url = get_absolute_url(user_message_path)
self.assertEqual(resp_data['user_message_url'], user_message_url)
# Verify that we were not enrolled
self.assertEqual(self._get_enrollments(), [])
@patch.dict(settings.FEATURES, {'EMBARGO': True})
def test_embargo_change_enrollment_restrict_geoip(self):
""" Validates that enrollment changes are blocked if the request originates from an embargoed country. """
# Use the helper to setup the embargo and simulate a request from a blocked IP address.
with restrict_course(self.course.id) as redirect_path:
self.assert_access_denied(redirect_path)
def _setup_embargo(self):
restricted_course = RestrictedCourse.objects.create(course_key=self.course.id)
restricted_country = Country.objects.create(country='US')
unrestricted_country = Country.objects.create(country='CA')
CountryAccessRule.objects.create(
rule_type=CountryAccessRule.BLACKLIST_RULE,
restricted_course=restricted_course,
country=restricted_country
)
# Clear the cache to remove the effects of previous embargo tests
cache.clear()
return unrestricted_country, restricted_country
@override_settings(EDX_API_KEY=EnrollmentTestMixin.API_KEY)
@patch.dict(settings.FEATURES, {'EMBARGO': True})
def test_embargo_change_enrollment_restrict_user_profile(self):
""" Validates that enrollment changes are blocked if the user's profile is linked to an embargoed country. """
__, restricted_country = self._setup_embargo()
# Update the user's profile, linking the user to the embargoed country.
self.user.profile.country = restricted_country.country
self.user.profile.save()
path = reverse('embargo:blocked_message', kwargs={'access_point': 'enrollment', 'message_key': 'default'})
self.assert_access_denied(path)
@override_settings(EDX_API_KEY=EnrollmentTestMixin.API_KEY)
@patch.dict(settings.FEATURES, {'EMBARGO': True})
def test_embargo_change_enrollment_allow_user_profile(self):
"""
Validates that enrollment changes are allowed if the user's profile is NOT linked to an embargoed country.
"""
# Setup the embargo
unrestricted_country, __ = self._setup_embargo()
# Verify that users without black-listed country codes *can* be enrolled
self.user.profile.country = unrestricted_country.country
self.user.profile.save()
self.assert_enrollment_status()
@patch.dict(settings.FEATURES, {'EMBARGO': True})
def test_embargo_change_enrollment_allow(self):
self.assert_enrollment_status()
# Verify that we were enrolled
self.assertEqual(len(self._get_enrollments()), 1)
def cross_domain_config(func):
"""Decorator for configuring a cross-domain request. """
feature_flag_decorator = patch.dict(settings.FEATURES, {
'ENABLE_CORS_HEADERS': True,
'ENABLE_CROSS_DOMAIN_CSRF_COOKIE': True
})
settings_decorator = override_settings(
CORS_ORIGIN_WHITELIST=["www.edx.org"],
CROSS_DOMAIN_CSRF_COOKIE_NAME="prod-edx-csrftoken",
CROSS_DOMAIN_CSRF_COOKIE_DOMAIN=".edx.org"
)
is_secure_decorator = patch.object(WSGIRequest, 'is_secure', return_value=True)
return feature_flag_decorator(
settings_decorator(
is_secure_decorator(func)
)
)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
class EnrollmentCrossDomainTest(ModuleStoreTestCase):
"""Test cross-domain calls to the enrollment end-points. """
USERNAME = "Bob"
EMAIL = "bob@example.com"
PASSWORD = "edx"
REFERER = "https://www.edx.org"
def setUp(self):
""" Create a course and user, then log in. """
super(EnrollmentCrossDomainTest, self).setUp()
self.course = CourseFactory.create()
self.user = UserFactory.create(username=self.USERNAME, email=self.EMAIL, password=self.PASSWORD)
self.client = Client(enforce_csrf_checks=True)
self.client.login(username=self.USERNAME, password=self.PASSWORD)
@cross_domain_config
def test_cross_domain_change_enrollment(self, *args): # pylint: disable=unused-argument
csrf_cookie = self._get_csrf_cookie()
resp = self._cross_domain_post(csrf_cookie)
# Expect that the request gets through successfully,
# passing the CSRF checks (including the referer check).
self.assertEqual(resp.status_code, 200)
@cross_domain_config
def test_cross_domain_missing_csrf(self, *args): # pylint: disable=unused-argument
resp = self._cross_domain_post('invalid_csrf_token')
self.assertEqual(resp.status_code, 403)
def _get_csrf_cookie(self):
"""Retrieve the cross-domain CSRF cookie. """
url = reverse('courseenrollment', kwargs={
'course_id': six.text_type(self.course.id)
})
resp = self.client.get(url, HTTP_REFERER=self.REFERER)
self.assertEqual(resp.status_code, 200)
self.assertIn('prod-edx-csrftoken', resp.cookies)
return resp.cookies['prod-edx-csrftoken'].value
def _cross_domain_post(self, csrf_cookie):
"""Perform a cross-domain POST request. """
url = reverse('courseenrollments')
params = json.dumps({
'course_details': {
'course_id': six.text_type(self.course.id),
},
'user': self.user.username
})
return self.client.post(
url, params, content_type='application/json',
HTTP_REFERER=self.REFERER,
HTTP_X_CSRFTOKEN=csrf_cookie
)
@ddt.ddt
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
class UnenrollmentTest(EnrollmentTestMixin, ModuleStoreTestCase):
"""
Tests unenrollment functionality. The API being tested is intended to
unenroll a learner from all of their courses.g
"""
USERNAME = "Bob"
EMAIL = "bob@example.com"
PASSWORD = "edx"
ENABLED_CACHES = ['default', 'mongo_metadata_inheritance', 'loc_cache']
ENABLED_SIGNALS = ['course_published']
def setUp(self):
""" Create a course and user, then log in. """
super(UnenrollmentTest, self).setUp()
self.superuser = SuperuserFactory()
# Pass emit_signals when creating the course so it would be cached
# as a CourseOverview. Enrollments require a cached CourseOverview.
self.first_org_course = CourseFactory.create(emit_signals=True, org="org", course="course", run="run")
self.other_first_org_course = CourseFactory.create(emit_signals=True, org="org", course="course2", run="run2")
self.second_org_course = CourseFactory.create(emit_signals=True, org="org2", course="course3", run="run3")
self.third_org_course = CourseFactory.create(emit_signals=True, org="org3", course="course4", run="run4")
self.courses = [
self.first_org_course, self.other_first_org_course, self.second_org_course, self.third_org_course
]
self.orgs = {"org", "org2", "org3"}
for course in self.courses:
CourseModeFactory.create(
course_id=str(course.id),
mode_slug=CourseMode.DEFAULT_MODE_SLUG,
mode_display_name=CourseMode.DEFAULT_MODE,
)
self.user = UserFactory.create(
username=self.USERNAME,
email=self.EMAIL,
password=self.PASSWORD,
)
self.client.login(username=self.USERNAME, password=self.PASSWORD)
for course in self.courses:
self.assert_enrollment_status(course_id=str(course.id), username=self.USERNAME, is_active=True)
def _create_test_retirement(self, user=None):
"""
Helper method to create a RetirementStatus with useful defaults
"""
RetirementState.objects.create(
state_name='PENDING',
state_execution_order=1,
is_dead_end_state=False,
required=False
)
if user is None:
user = UserFactory()
return UserRetirementStatus.create_retirement(user)
def build_jwt_headers(self, user):
"""
Helper function for creating headers for the JWT authentication.
"""
token = create_jwt_for_user(user)
headers = {'HTTP_AUTHORIZATION': 'JWT ' + token}
return headers
def test_deactivate_enrollments(self):
self._assert_active()
self._create_test_retirement(self.user)
response = self._submit_unenroll(self.superuser, self.user.username)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = json.loads(response.content)
# order doesn't matter so compare sets
self.assertEqual(set(data), self.orgs)
self._assert_inactive()
def test_deactivate_enrollments_no_retirement_status(self):
self._assert_active()
response = self._submit_unenroll(self.superuser, self.user.username)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
def test_deactivate_enrollments_unauthorized(self):
self._assert_active()
response = self._submit_unenroll(self.user, self.user.username)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self._assert_active()
def test_deactivate_enrollments_no_username(self):
self._assert_active()
response = self._submit_unenroll(self.superuser, None)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
data = json.loads(response.content)
self.assertEqual(data, u"Username not specified.")
self._assert_active()
def test_deactivate_enrollments_empty_username(self):
self._assert_active()
self._create_test_retirement(self.user)
response = self._submit_unenroll(self.superuser, "")
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self._assert_active()
def test_deactivate_enrollments_invalid_username(self):
self._assert_active()
self._create_test_retirement(self.user)
response = self._submit_unenroll(self.superuser, "a made up username")
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self._assert_active()
def test_deactivate_enrollments_called_twice(self):
self._assert_active()
self._create_test_retirement(self.user)
response = self._submit_unenroll(self.superuser, self.user.username)
self.assertEqual(response.status_code, status.HTTP_200_OK)
response = self._submit_unenroll(self.superuser, self.user.username)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.assertEqual(response.content, "")
self._assert_inactive()
def _assert_active(self):
for course in self.courses:
self.assertTrue(CourseEnrollment.is_enrolled(self.user, course.id))
_, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, course.id)
self.assertTrue(is_active)
def _assert_inactive(self):
for course in self.courses:
_, is_active = CourseEnrollment.enrollment_mode_for_user(self.user, course.id)
self.assertFalse(is_active)
def _submit_unenroll(self, submitting_user, unenrolling_username):
data = {}
if unenrolling_username is not None:
data['username'] = unenrolling_username
url = reverse('unenrollment')
headers = self.build_jwt_headers(submitting_user)
return self.client.post(url, json.dumps(data), content_type='application/json', **headers)
@ddt.ddt
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
class UserRoleTest(ModuleStoreTestCase):
"""
Tests the API call to list user roles.
"""
USERNAME = "Bob"
EMAIL = "bob@example.com"
STAFF_USERNAME = "Bobstaff"
STAFF_EMAIL = "bobStaff@example.com"
PASSWORD = "edx"
ENABLED_CACHES = ['default']
def setUp(self):
""" Create a course and user, then log in. """
super(UserRoleTest, self).setUp()
self.course1 = CourseFactory.create(emit_signals=True, org="org1", course="course1", run="run1")
self.course2 = CourseFactory.create(emit_signals=True, org="org2", course="course2", run="run2")
self.user = UserFactory.create(
username=self.USERNAME,
email=self.EMAIL,
password=self.PASSWORD,
)
self.staff_user = UserFactory.create(
username=self.STAFF_USERNAME,
email=self.STAFF_EMAIL,
password=self.PASSWORD,
is_staff=True,
)
self.client.login(username=self.USERNAME, password=self.PASSWORD)
def _create_expected_role_dict(self, course, role):
""" Creates the expected role dict object that the view should return """
return {
'course_id': text_type(course.id),
'org': course.org,
'role': role.ROLE,
}
def _assert_roles(self, expected_roles, is_staff, course_id=None):
""" Asserts that the api call is successful and returns the expected roles """
if course_id is not None:
response = self.client.get(reverse('roles'), {'course_id': course_id})
else:
response = self.client.get(reverse('roles'))
self.assertEqual(response.status_code, status.HTTP_200_OK)
response_data = json.loads(response.content)
sort_by_role_id = lambda r: r['course_id']
response_data['roles'] = sorted(response_data['roles'], key=sort_by_role_id)
expected_roles = sorted(expected_roles, key=sort_by_role_id)
expected = {'roles': expected_roles, 'is_staff': is_staff}
self.assertEqual(response_data, expected)
def _login(self, is_staff):
""" If is_staff is true, logs in the staff user. Otherwise, logs in the non-staff user """
logged_in_user = self.staff_user if is_staff else self.user
self.client.login(username=logged_in_user.username, password=self.PASSWORD)
return logged_in_user
def test_not_logged_in(self):
self.client.logout()
response = self.client.get(reverse('roles'))
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
@ddt.data(True, False)
def test_roles_no_roles(self, is_staff):
self._login(is_staff)
self._assert_roles([], is_staff)
@ddt.data(True, False)
def test_roles(self, is_staff):
logged_in_user = self._login(is_staff)
role1 = CourseStaffRole(self.course1.id)
role1.add_users(logged_in_user)
expected_role1 = self._create_expected_role_dict(self.course1, role1)
expected_roles = [expected_role1]
self._assert_roles(expected_roles, is_staff)
role2 = CourseStaffRole(self.course2.id)
role2.add_users(logged_in_user)
expected_role2 = self._create_expected_role_dict(self.course2, role2)
expected_roles.append(expected_role2)
self._assert_roles(expected_roles, is_staff)
def test_roles_filter(self):
role1 = CourseStaffRole(self.course1.id)
role1.add_users(self.user)
expected_role1 = self._create_expected_role_dict(self.course1, role1)
role2 = CourseStaffRole(self.course2.id)
role2.add_users(self.user)
expected_role2 = self._create_expected_role_dict(self.course2, role2)
self._assert_roles([expected_role1], False, course_id=text_type(self.course1.id))
self._assert_roles([expected_role2], False, course_id=text_type(self.course2.id))
def test_roles_exception(self):
with patch('openedx.core.djangoapps.enrollments.api.get_user_roles') as mock_get_user_roles:
mock_get_user_roles.side_effect = Exception()
response = self.client.get(reverse('roles'))
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
expected_response = {
"message": (
u"An error occurred while retrieving roles for user '{username}"
).format(username=self.user.username)
}
response_data = json.loads(response.content)
self.assertEqual(response_data, expected_response)
@ddt.ddt
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
class CourseEnrollmentsApiListTest(APITestCase, ModuleStoreTestCase):
"""
Test the course enrollments list API.
"""
CREATED_DATA = datetime.datetime(2018, 1, 1, 0, 0, 1, tzinfo=pytz.UTC)
def setUp(self):
super(CourseEnrollmentsApiListTest, self).setUp()
self.rate_limit_config = RateLimitConfiguration.current()
self.rate_limit_config.enabled = False
self.rate_limit_config.save()
throttle = EnrollmentUserThrottle()
self.rate_limit, __ = throttle.parse_rate(throttle.rate)
self.course = CourseFactory.create(org='e', number='d', run='X', emit_signals=True)
self.course2 = CourseFactory.create(org='x', number='y', run='Z', emit_signal=True)
for mode_slug in ('honor', 'verified', 'audit'):
CourseModeFactory.create(
course_id=self.course.id,
mode_slug=mode_slug,
mode_display_name=mode_slug
)
self.staff_user = AdminFactory(
username='staff',
email='staff@example.com',
password='edx'
)
self.student1 = UserFactory(
username='student1',
email='student1@example.com',
password='edx'
)
self.student2 = UserFactory(
username='student2',
email='student2@example.com',
password='edx'
)
self.student3 = UserFactory(
username='student3',
email='student3@example.com',
password='edx'
)
with freeze_time(self.CREATED_DATA):
data.create_course_enrollment(
self.student1.username,
six.text_type(self.course.id),
'honor',
True
)
data.create_course_enrollment(
self.student2.username,
six.text_type(self.course.id),
'honor',
True
)
data.create_course_enrollment(
self.student3.username,
six.text_type(self.course2.id),
'verified',
True
)
data.create_course_enrollment(
self.student2.username,
six.text_type(self.course2.id),
'honor',
True
)
data.create_course_enrollment(
self.staff_user.username,
six.text_type(self.course2.id),
'verified',
True
)
self.url = reverse('courseenrollmentsapilist')
def _login_as_staff(self):
self.client.login(username=self.staff_user.username, password='edx')
def _make_request(self, query_params=None):
return self.client.get(self.url, query_params)
def _assert_list_of_enrollments(self, query_params=None, expected_status=status.HTTP_200_OK, error_fields=None):
"""
Make a request to the CourseEnrolllmentApiList endpoint and run assertions on the response
using the optional parameters 'query_params', 'expected_status' and 'error_fields'.
"""
response = self._make_request(query_params)
self.assertEqual(response.status_code, expected_status)
content = json.loads(response.content)
if error_fields is not None:
self.assertIn('field_errors', content)
for error_field in error_fields:
self.assertIn(error_field, content['field_errors'])
return content
def test_user_not_authenticated(self):
self.client.logout()
response = self.client.get(self.url, {'course_id': self.course.id})
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
def test_user_not_authorized(self):
self.client.login(username=self.student1.username, password='edx')
response = self.client.get(self.url, {'course_id': self.course.id})
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
@ddt.data(
({'course_id': '1'}, ['course_id', ]),
({'course_id': '1', 'username': 'staff'}, ['course_id', ]),
({'username': '1*2'}, ['username', ]),
({'username': '1*2', 'course_id': 'org.0/course_0/Run_0'}, ['username', ]),
({'username': '1*2', 'course_id': '1'}, ['username', 'course_id']),
({'username': ','.join(str(x) for x in range(101))}, ['username', ])
)
@ddt.unpack
def test_query_string_parameters_invalid_errors(self, query_params, error_fields):
self._login_as_staff()
self._assert_list_of_enrollments(query_params, status.HTTP_400_BAD_REQUEST, error_fields)
@ddt.data(
# Non-existent user
({'username': 'nobody'}, ),
({'username': 'nobody', 'course_id': 'e/d/X'}, ),
# Non-existent course
({'course_id': 'a/b/c'}, ),
({'course_id': 'a/b/c', 'username': 'student1'}, ),
# Non-existent course and user
({'course_id': 'a/b/c', 'username': 'dummy'}, )
)
@ddt.unpack
def test_non_existent_course_user(self, query_params):
self._login_as_staff()
content = self._assert_list_of_enrollments(query_params, status.HTTP_200_OK)
self.assertEqual(len(content['results']), 0)
@ddt.file_data('fixtures/course-enrollments-api-list-valid-data.json')
@ddt.unpack
def test_response_valid_queries(self, args):
query_params = args[0]
expected_results = args[1]
self._login_as_staff()
content = self._assert_list_of_enrollments(query_params, status.HTTP_200_OK)
results = content['results']
self.assertItemsEqual(results, expected_results)

View File

@@ -0,0 +1,32 @@
"""
URLs for the Enrollment API
"""
from __future__ import absolute_import
from django.conf import settings
from django.conf.urls import url
from .views import (
CourseEnrollmentsApiListView,
EnrollmentCourseDetailView,
EnrollmentListView,
EnrollmentUserRolesView,
EnrollmentView,
UnenrollmentView
)
urlpatterns = [
url(r'^enrollment/{username},{course_key}$'.format(
username=settings.USERNAME_PATTERN,
course_key=settings.COURSE_ID_PATTERN),
EnrollmentView.as_view(), name='courseenrollment'),
url(r'^enrollment/{course_key}$'.format(course_key=settings.COURSE_ID_PATTERN),
EnrollmentView.as_view(), name='courseenrollment'),
url(r'^enrollment$', EnrollmentListView.as_view(), name='courseenrollments'),
url(r'^enrollments/?$', CourseEnrollmentsApiListView.as_view(), name='courseenrollmentsapilist'),
url(r'^course/{course_key}$'.format(course_key=settings.COURSE_ID_PATTERN),
EnrollmentCourseDetailView.as_view(), name='courseenrollmentdetails'),
url(r'^unenroll/$', UnenrollmentView.as_view(), name='unenrollment'),
url(r'^roles/$', EnrollmentUserRolesView.as_view(), name='roles'),
]

View File

@@ -0,0 +1,966 @@
"""
The Enrollment API Views should be simple, lean HTTP endpoints for API access. This should
consist primarily of authentication, request validation, and serialization.
"""
from __future__ import absolute_import
import logging
from six import text_type
from course_modes.models import CourseMode
from django.core.exceptions import ObjectDoesNotExist, ValidationError
from django.utils.decorators import method_decorator
from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
from edx_rest_framework_extensions.auth.session.authentication import SessionAuthenticationAllowInactiveUser
from opaque_keys import InvalidKeyError
from opaque_keys.edx.keys import CourseKey
from openedx.core.djangoapps.cors_csrf.authentication import SessionAuthenticationCrossDomainCsrf
from openedx.core.djangoapps.cors_csrf.decorators import ensure_csrf_cookie_cross_domain
from openedx.core.djangoapps.course_groups.cohorts import CourseUserGroup, add_user_to_cohort, get_cohort_by_name
from openedx.core.djangoapps.embargo import api as embargo_api
from openedx.core.djangoapps.enrollments import api
from openedx.core.djangoapps.enrollments.errors import (
CourseEnrollmentError, CourseEnrollmentExistsError, CourseModeNotFoundError,
)
from openedx.core.djangoapps.enrollments.forms import CourseEnrollmentsApiListForm
from openedx.core.djangoapps.enrollments.paginators import CourseEnrollmentsApiListPagination
from openedx.core.djangoapps.enrollments.serializers import CourseEnrollmentsApiListSerializer
from openedx.core.djangoapps.user_api.accounts.permissions import CanRetireUser
from openedx.core.djangoapps.user_api.models import UserRetirementStatus
from openedx.core.djangoapps.user_api.preferences.api import update_email_opt_in
from openedx.core.lib.api.authentication import OAuth2AuthenticationAllowInactiveUser
from openedx.core.lib.api.permissions import ApiKeyHeaderPermission, ApiKeyHeaderPermissionIsAuthenticated
from openedx.core.lib.api.view_utils import DeveloperErrorViewMixin
from openedx.core.lib.exceptions import CourseNotFoundError
from openedx.core.lib.log_utils import audit_log
from openedx.features.enterprise_support.api import (
ConsentApiServiceClient,
EnterpriseApiException,
EnterpriseApiServiceClient,
enterprise_enabled
)
from rest_framework import permissions, status
from rest_framework.generics import ListAPIView
from rest_framework.response import Response
from rest_framework.throttling import UserRateThrottle
from rest_framework.views import APIView
from student.auth import user_has_role
from student.models import CourseEnrollment, User
from student.roles import CourseStaffRole, GlobalStaff
from util.disable_rate_limit import can_disable_rate_limit
log = logging.getLogger(__name__)
REQUIRED_ATTRIBUTES = {
"credit": ["credit:provider_id"],
}
class EnrollmentCrossDomainSessionAuth(SessionAuthenticationAllowInactiveUser, SessionAuthenticationCrossDomainCsrf):
"""Session authentication that allows inactive users and cross-domain requests. """
pass
class ApiKeyPermissionMixIn(object):
"""
This mixin is used to provide a convenience function for doing individual permission checks
for the presence of API keys.
"""
def has_api_key_permissions(self, request):
"""
Checks to see if the request was made by a server with an API key.
Args:
request (Request): the request being made into the view
Return:
True if the request has been made with a valid API key
False otherwise
"""
return ApiKeyHeaderPermission().has_permission(request, self)
class EnrollmentUserThrottle(UserRateThrottle, ApiKeyPermissionMixIn):
"""Limit the number of requests users can make to the enrollment API."""
# To see how the staff rate limit was selected, see https://github.com/edx/edx-platform/pull/18360
THROTTLE_RATES = {
'user': '40/minute',
'staff': '120/minute',
}
def allow_request(self, request, view):
# Use a special scope for staff to allow for a separate throttle rate
user = request.user
if user.is_authenticated and (user.is_staff or user.is_superuser):
self.scope = 'staff'
self.rate = self.get_rate()
self.num_requests, self.duration = self.parse_rate(self.rate)
return self.has_api_key_permissions(request) or super(EnrollmentUserThrottle, self).allow_request(request, view)
@can_disable_rate_limit
class EnrollmentView(APIView, ApiKeyPermissionMixIn):
"""
**Use Case**
Get the user's enrollment status for a course.
**Example Request**
GET /api/enrollment/v1/enrollment/{username},{course_id}
**Response Values**
If the request for information about the user is successful, an HTTP 200 "OK" response
is returned.
The HTTP 200 response has the following values.
* course_details: A collection that includes the following
values.
* course_end: The date and time when the course closes. If
null, the course never ends.
* course_id: The unique identifier for the course.
* course_name: The name of the course.
* course_modes: An array of data about the enrollment modes
supported for the course. If the request uses the parameter
include_expired=1, the array also includes expired
enrollment modes.
Each enrollment mode collection includes the following
values.
* currency: The currency of the listed prices.
* description: A description of this mode.
* expiration_datetime: The date and time after which
users cannot enroll in the course in this mode.
* min_price: The minimum price for which a user can
enroll in this mode.
* name: The full name of the enrollment mode.
* slug: The short name for the enrollment mode.
* suggested_prices: A list of suggested prices for
this enrollment mode.
* course_end: The date and time at which the course closes. If
null, the course never ends.
* course_start: The date and time when the course opens. If
null, the course opens immediately when it is created.
* enrollment_end: The date and time after which users cannot
enroll for the course. If null, the enrollment period never
ends.
* enrollment_start: The date and time when users can begin
enrolling in the course. If null, enrollment opens
immediately when the course is created.
* invite_only: A value indicating whether students must be
invited to enroll in the course. Possible values are true or
false.
* created: The date the user account was created.
* is_active: Whether the enrollment is currently active.
* mode: The enrollment mode of the user in this course.
* user: The ID of the user.
"""
authentication_classes = (
JwtAuthentication,
OAuth2AuthenticationAllowInactiveUser,
SessionAuthenticationAllowInactiveUser,
)
permission_classes = (ApiKeyHeaderPermissionIsAuthenticated,)
throttle_classes = (EnrollmentUserThrottle,)
# Since the course about page on the marketing site uses this API to auto-enroll users,
# we need to support cross-domain CSRF.
@method_decorator(ensure_csrf_cookie_cross_domain)
def get(self, request, course_id=None, username=None):
"""Create, read, or update enrollment information for a user.
HTTP Endpoint for all CRUD operations for a user course enrollment. Allows creation, reading, and
updates of the current enrollment for a particular course.
Args:
request (Request): To get current course enrollment information, a GET request will return
information for the current user and the specified course.
course_id (str): URI element specifying the course location. Enrollment information will be
returned, created, or updated for this particular course.
username (str): The username associated with this enrollment request.
Return:
A JSON serialized representation of the course enrollment.
"""
username = username or request.user.username
# TODO Implement proper permissions
if request.user.username != username and not self.has_api_key_permissions(request) \
and not request.user.is_staff:
# Return a 404 instead of a 403 (Unauthorized). If one user is looking up
# other users, do not let them deduce the existence of an enrollment.
return Response(status=status.HTTP_404_NOT_FOUND)
try:
return Response(api.get_enrollment(username, course_id))
except CourseEnrollmentError:
return Response(
status=status.HTTP_400_BAD_REQUEST,
data={
"message": (
u"An error occurred while retrieving enrollments for user "
u"'{username}' in course '{course_id}'"
).format(username=username, course_id=course_id)
}
)
class EnrollmentUserRolesView(APIView):
"""
**Use Case**
Get the roles for the current logged-in user.
A field is also included to indicate whether or not the user is a global
staff member.
If an optional course_id parameter is supplied, the returned roles will be
filtered to only include roles for the given course.
**Example Requests**
GET /api/enrollment/v1/roles/?course_id={course_id}
course_id: (optional) A course id. The returned roles will be filtered to
only include roles for the given course.
**Response Values**
If the request is successful, an HTTP 200 "OK" response is
returned along with a collection of user roles for the
logged-in user, filtered by course_id if given, along with
whether or not the user is global staff
"""
authentication_classes = (
JwtAuthentication,
OAuth2AuthenticationAllowInactiveUser,
EnrollmentCrossDomainSessionAuth,
)
permission_classes = (ApiKeyHeaderPermissionIsAuthenticated,)
throttle_classes = (EnrollmentUserThrottle,)
@method_decorator(ensure_csrf_cookie_cross_domain)
def get(self, request):
"""
Gets a list of all roles for the currently logged-in user, filtered by course_id if supplied
"""
try:
course_id = request.GET.get('course_id')
roles_data = api.get_user_roles(request.user.username)
if course_id:
roles_data = [role for role in roles_data if text_type(role.course_id) == course_id]
except Exception: # pylint: disable=broad-except
return Response(
status=status.HTTP_400_BAD_REQUEST,
data={
"message": (
u"An error occurred while retrieving roles for user '{username}"
).format(username=request.user.username)
}
)
return Response({
'roles': [
{
"org": role.org,
"course_id": text_type(role.course_id),
"role": role.role
}
for role in roles_data],
'is_staff': request.user.is_staff,
})
@can_disable_rate_limit
class EnrollmentCourseDetailView(APIView):
"""
**Use Case**
Get enrollment details for a course.
Response values include the course schedule and enrollment modes
supported by the course. Use the parameter include_expired=1 to
include expired enrollment modes in the response.
**Note:** Getting enrollment details for a course does not require
authentication.
**Example Requests**
GET /api/enrollment/v1/course/{course_id}
GET /api/enrollment/v1/course/{course_id}?include_expired=1
**Response Values**
If the request is successful, an HTTP 200 "OK" response is
returned along with a collection of course enrollments for the
user or for the newly created enrollment.
Each course enrollment contains the following values.
* course_end: The date and time when the course closes. If
null, the course never ends.
* course_id: The unique identifier for the course.
* course_name: The name of the course.
* course_modes: An array of data about the enrollment modes
supported for the course. If the request uses the parameter
include_expired=1, the array also includes expired
enrollment modes.
Each enrollment mode collection includes the following
values.
* currency: The currency of the listed prices.
* description: A description of this mode.
* expiration_datetime: The date and time after which
users cannot enroll in the course in this mode.
* min_price: The minimum price for which a user can
enroll in this mode.
* name: The full name of the enrollment mode.
* slug: The short name for the enrollment mode.
* suggested_prices: A list of suggested prices for
this enrollment mode.
* course_start: The date and time when the course opens. If
null, the course opens immediately when it is created.
* enrollment_end: The date and time after which users cannot
enroll for the course. If null, the enrollment period never
ends.
* enrollment_start: The date and time when users can begin
enrolling in the course. If null, enrollment opens
immediately when the course is created.
* invite_only: A value indicating whether students must be
invited to enroll in the course. Possible values are true or
false.
"""
authentication_classes = []
permission_classes = []
throttle_classes = (EnrollmentUserThrottle,)
def get(self, request, course_id=None):
"""Read enrollment information for a particular course.
HTTP Endpoint for retrieving course level enrollment information.
Args:
request (Request): To get current course enrollment information, a GET request will return
information for the specified course.
course_id (str): URI element specifying the course location. Enrollment information will be
returned.
Return:
A JSON serialized representation of the course enrollment details.
"""
try:
return Response(api.get_course_enrollment_details(course_id, bool(request.GET.get('include_expired', ''))))
except CourseNotFoundError:
return Response(
status=status.HTTP_400_BAD_REQUEST,
data={
"message": (
u"No course found for course ID '{course_id}'"
).format(course_id=course_id)
}
)
class UnenrollmentView(APIView):
"""
**Use Cases**
* Unenroll a single user from all courses.
This command can only be issued by a privileged service user.
**Example Requests**
POST /api/enrollment/v1/enrollment {
"username": "username12345"
}
**POST Parameters**
A POST request must include the following parameter.
* username: The username of the user being unenrolled.
This will never match the username from the request,
since the request is issued as a privileged service user.
**POST Response Values**
If the user has not requested retirement and does not have a retirement
request status, the request returns an HTTP 404 "Does Not Exist" response.
If the user is already unenrolled from all courses, the request returns
an HTTP 204 "No Content" response.
If an unexpected error occurs, the request returns an HTTP 500 response.
If the request is successful, an HTTP 200 "OK" response is
returned along with a list of all courses from which the user was unenrolled.
"""
authentication_classes = (JwtAuthentication,)
permission_classes = (permissions.IsAuthenticated, CanRetireUser,)
def post(self, request):
"""
Unenrolls the specified user from all courses.
"""
try:
# Get the username from the request.
username = request.data['username']
# Ensure that a retirement request status row exists for this username.
UserRetirementStatus.get_retirement_for_retirement_action(username)
enrollments = api.get_enrollments(username)
active_enrollments = [enrollment for enrollment in enrollments if enrollment['is_active']]
if len(active_enrollments) < 1:
return Response(status=status.HTTP_204_NO_CONTENT)
return Response(api.unenroll_user_from_all_courses(username))
except KeyError:
return Response(u'Username not specified.', status=status.HTTP_404_NOT_FOUND)
except UserRetirementStatus.DoesNotExist:
return Response(u'No retirement request status for username.', status=status.HTTP_404_NOT_FOUND)
except Exception as exc: # pylint: disable=broad-except
return Response(text_type(exc), status=status.HTTP_500_INTERNAL_SERVER_ERROR)
@can_disable_rate_limit
class EnrollmentListView(APIView, ApiKeyPermissionMixIn):
"""
**Use Cases**
* Get a list of all course enrollments for the currently signed in user.
* Enroll the currently signed in user in a course.
Currently a user can use this command only to enroll the
user in the default course mode. If this is not
supported for the course, the request fails and returns
the available modes.
This command can use a server-to-server call to enroll a user in
other modes, such as "verified", "professional", or "credit". If
the mode is not supported for the course, the request will fail
and return the available modes.
You can include other parameters as enrollment attributes for a
specific course mode. For example, for credit mode, you can
include the following parameters to specify the credit provider
attribute.
* namespace: credit
* name: provider_id
* value: institution_name
**Example Requests**
GET /api/enrollment/v1/enrollment
POST /api/enrollment/v1/enrollment {
"mode": "credit",
"course_details":{"course_id": "edX/DemoX/Demo_Course"},
"enrollment_attributes":[{"namespace": "credit","name": "provider_id","value": "hogwarts",},]
}
**POST Parameters**
A POST request can include the following parameters.
* user: Optional. The username of the currently logged in user.
You cannot use the command to enroll a different user.
* mode: Optional. The course mode for the enrollment. Individual
users cannot upgrade their enrollment mode from the default. Only
server-to-server requests can enroll with other modes.
* is_active: Optional. A Boolean value indicating whether the
enrollment is active. Only server-to-server requests are
allowed to deactivate an enrollment.
* course details: A collection that includes the following
information.
* course_id: The unique identifier for the course.
* email_opt_in: Optional. A Boolean value that indicates whether
the user wants to receive email from the organization that runs
this course.
* enrollment_attributes: A dictionary that contains the following
values.
* namespace: Namespace of the attribute
* name: Name of the attribute
* value: Value of the attribute
* is_active: Optional. A Boolean value that indicates whether the
enrollment is active. Only server-to-server requests can
deactivate an enrollment.
* mode: Optional. The course mode for the enrollment. Individual
users cannot upgrade their enrollment mode from the default. Only
server-to-server requests can enroll with other modes.
* user: Optional. The user ID of the currently logged in user. You
cannot use the command to enroll a different user.
* enterprise_course_consent: Optional. A Boolean value that
indicates the consent status for an EnterpriseCourseEnrollment
to be posted to the Enterprise service.
**GET Response Values**
If an unspecified error occurs when the user tries to obtain a
learner's enrollments, the request returns an HTTP 400 "Bad
Request" response.
If the user does not have permission to view enrollment data for
the requested learner, the request returns an HTTP 404 "Not Found"
response.
**POST Response Values**
If the user does not specify a course ID, the specified course
does not exist, or the is_active status is invalid, the request
returns an HTTP 400 "Bad Request" response.
If a user who is not an admin tries to upgrade a learner's course
mode, the request returns an HTTP 403 "Forbidden" response.
If the specified user does not exist, the request returns an HTTP
406 "Not Acceptable" response.
**GET and POST Response Values**
If the request is successful, an HTTP 200 "OK" response is
returned along with a collection of course enrollments for the
user or for the newly created enrollment.
Each course enrollment contains the following values.
* course_details: A collection that includes the following
values.
* course_end: The date and time when the course closes. If
null, the course never ends.
* course_id: The unique identifier for the course.
* course_name: The name of the course.
* course_modes: An array of data about the enrollment modes
supported for the course. If the request uses the parameter
include_expired=1, the array also includes expired
enrollment modes.
Each enrollment mode collection includes the following
values.
* currency: The currency of the listed prices.
* description: A description of this mode.
* expiration_datetime: The date and time after which users
cannot enroll in the course in this mode.
* min_price: The minimum price for which a user can enroll in
this mode.
* name: The full name of the enrollment mode.
* slug: The short name for the enrollment mode.
* suggested_prices: A list of suggested prices for this
enrollment mode.
* course_start: The date and time when the course opens. If
null, the course opens immediately when it is created.
* enrollment_end: The date and time after which users cannot
enroll for the course. If null, the enrollment period never
ends.
* enrollment_start: The date and time when users can begin
enrolling in the course. If null, enrollment opens
immediately when the course is created.
* invite_only: A value indicating whether students must be
invited to enroll in the course. Possible values are true or
false.
* created: The date the user account was created.
* is_active: Whether the enrollment is currently active.
* mode: The enrollment mode of the user in this course.
* user: The username of the user.
"""
authentication_classes = (
JwtAuthentication,
OAuth2AuthenticationAllowInactiveUser,
EnrollmentCrossDomainSessionAuth,
)
permission_classes = (ApiKeyHeaderPermissionIsAuthenticated,)
throttle_classes = (EnrollmentUserThrottle,)
# Since the course about page on the marketing site
# uses this API to auto-enroll users, we need to support
# cross-domain CSRF.
@method_decorator(ensure_csrf_cookie_cross_domain)
def get(self, request):
"""Gets a list of all course enrollments for a user.
Returns a list for the currently logged in user, or for the user named by the 'user' GET
parameter. If the username does not match that of the currently logged in user, only
courses for which the currently logged in user has the Staff or Admin role are listed.
As a result, a course team member can find out which of his or her own courses a particular
learner is enrolled in.
Only the Staff or Admin role (granted on the Django administrative console as the staff
or instructor permission) in individual courses gives the requesting user access to
enrollment data. Permissions granted at the organizational level do not give a user
access to enrollment data for all of that organization's courses.
Users who have the global staff permission can access all enrollment data for all
courses.
"""
username = request.GET.get('user', request.user.username)
try:
enrollment_data = api.get_enrollments(username)
except CourseEnrollmentError:
return Response(
status=status.HTTP_400_BAD_REQUEST,
data={
"message": (
u"An error occurred while retrieving enrollments for user '{username}'"
).format(username=username)
}
)
if username == request.user.username or GlobalStaff().has_user(request.user) or \
self.has_api_key_permissions(request):
return Response(enrollment_data)
filtered_data = []
for enrollment in enrollment_data:
course_key = CourseKey.from_string(enrollment["course_details"]["course_id"])
if user_has_role(request.user, CourseStaffRole(course_key)):
filtered_data.append(enrollment)
return Response(filtered_data)
def post(self, request):
# pylint: disable=too-many-statements
"""Enrolls the currently logged-in user in a course.
Server-to-server calls may deactivate or modify the mode of existing enrollments. All other requests
go through `add_enrollment()`, which allows creation of new and reactivation of old enrollments.
"""
# Get the User, Course ID, and Mode from the request.
username = request.data.get('user', request.user.username)
course_id = request.data.get('course_details', {}).get('course_id')
if not course_id:
return Response(
status=status.HTTP_400_BAD_REQUEST,
data={"message": u"Course ID must be specified to create a new enrollment."}
)
try:
course_id = CourseKey.from_string(course_id)
except InvalidKeyError:
return Response(
status=status.HTTP_400_BAD_REQUEST,
data={
"message": u"No course '{course_id}' found for enrollment".format(course_id=course_id)
}
)
mode = request.data.get('mode')
has_api_key_permissions = self.has_api_key_permissions(request)
# Check that the user specified is either the same user, or this is a server-to-server request.
if not username:
username = request.user.username
if username != request.user.username and not has_api_key_permissions:
# Return a 404 instead of a 403 (Unauthorized). If one user is looking up
# other users, do not let them deduce the existence of an enrollment.
return Response(status=status.HTTP_404_NOT_FOUND)
if mode not in (CourseMode.AUDIT, CourseMode.HONOR, None) and not has_api_key_permissions:
return Response(
status=status.HTTP_403_FORBIDDEN,
data={
"message": u"User does not have permission to create enrollment with mode [{mode}].".format(
mode=mode
)
}
)
try:
# Lookup the user, instead of using request.user, since request.user may not match the username POSTed.
user = User.objects.get(username=username)
except ObjectDoesNotExist:
return Response(
status=status.HTTP_406_NOT_ACCEPTABLE,
data={
'message': u'The user {} does not exist.'.format(username)
}
)
embargo_response = embargo_api.get_embargo_response(request, course_id, user)
if embargo_response:
return embargo_response
try:
is_active = request.data.get('is_active')
# Check if the requested activation status is None or a Boolean
if is_active is not None and not isinstance(is_active, bool):
return Response(
status=status.HTTP_400_BAD_REQUEST,
data={
'message': (u"'{value}' is an invalid enrollment activation status.").format(value=is_active)
}
)
explicit_linked_enterprise = request.data.get('linked_enterprise_customer')
if explicit_linked_enterprise and has_api_key_permissions and enterprise_enabled():
enterprise_api_client = EnterpriseApiServiceClient()
consent_client = ConsentApiServiceClient()
try:
enterprise_api_client.post_enterprise_course_enrollment(username, text_type(course_id), None)
except EnterpriseApiException as error:
log.exception(u"An unexpected error occurred while creating the new EnterpriseCourseEnrollment "
u"for user [%s] in course run [%s]", username, course_id)
raise CourseEnrollmentError(text_type(error))
kwargs = {
'username': username,
'course_id': text_type(course_id),
'enterprise_customer_uuid': explicit_linked_enterprise,
}
consent_client.provide_consent(**kwargs)
enrollment_attributes = request.data.get('enrollment_attributes')
enrollment = api.get_enrollment(username, text_type(course_id))
mode_changed = enrollment and mode is not None and enrollment['mode'] != mode
active_changed = enrollment and is_active is not None and enrollment['is_active'] != is_active
missing_attrs = []
if enrollment_attributes:
actual_attrs = [
u"{namespace}:{name}".format(**attr)
for attr in enrollment_attributes
]
missing_attrs = set(REQUIRED_ATTRIBUTES.get(mode, [])) - set(actual_attrs)
if has_api_key_permissions and (mode_changed or active_changed):
if mode_changed and active_changed and not is_active:
# if the requester wanted to deactivate but specified the wrong mode, fail
# the request (on the assumption that the requester had outdated information
# about the currently active enrollment).
msg = u"Enrollment mode mismatch: active mode={}, requested mode={}. Won't deactivate.".format(
enrollment["mode"], mode
)
log.warning(msg)
return Response(status=status.HTTP_400_BAD_REQUEST, data={"message": msg})
if missing_attrs:
msg = u"Missing enrollment attributes: requested mode={} required attributes={}".format(
mode, REQUIRED_ATTRIBUTES.get(mode)
)
log.warning(msg)
return Response(status=status.HTTP_400_BAD_REQUEST, data={"message": msg})
response = api.update_enrollment(
username,
text_type(course_id),
mode=mode,
is_active=is_active,
enrollment_attributes=enrollment_attributes,
# If we are updating enrollment by authorized api caller, we should allow expired modes
include_expired=has_api_key_permissions
)
else:
# Will reactivate inactive enrollments.
response = api.add_enrollment(
username,
text_type(course_id),
mode=mode,
is_active=is_active,
enrollment_attributes=enrollment_attributes
)
cohort_name = request.data.get('cohort')
if cohort_name is not None:
cohort = get_cohort_by_name(course_id, cohort_name)
try:
add_user_to_cohort(cohort, user)
except ValueError:
# user already in cohort, probably because they were un-enrolled and re-enrolled
log.exception('Cohort re-addition')
email_opt_in = request.data.get('email_opt_in', None)
if email_opt_in is not None:
org = course_id.org
update_email_opt_in(request.user, org, email_opt_in)
log.info(u'The user [%s] has already been enrolled in course run [%s].', username, course_id)
return Response(response)
except CourseModeNotFoundError as error:
return Response(
status=status.HTTP_400_BAD_REQUEST,
data={
"message": (
u"The [{mode}] course mode is expired or otherwise unavailable for course run [{course_id}]."
).format(mode=mode, course_id=course_id),
"course_details": error.data
})
except CourseNotFoundError:
return Response(
status=status.HTTP_400_BAD_REQUEST,
data={
"message": u"No course '{course_id}' found for enrollment".format(course_id=course_id)
}
)
except CourseEnrollmentExistsError as error:
log.warning(u'An enrollment already exists for user [%s] in course run [%s].', username, course_id)
return Response(data=error.enrollment)
except CourseEnrollmentError:
log.exception(u"An error occurred while creating the new course enrollment for user "
u"[%s] in course run [%s]", username, course_id)
return Response(
status=status.HTTP_400_BAD_REQUEST,
data={
"message": (
u"An error occurred while creating the new course enrollment for user "
u"'{username}' in course '{course_id}'"
).format(username=username, course_id=course_id)
}
)
except CourseUserGroup.DoesNotExist:
log.exception(u'Missing cohort [%s] in course run [%s]', cohort_name, course_id)
return Response(
status=status.HTTP_400_BAD_REQUEST,
data={
"message": u"An error occured while adding to cohort [%s]" % cohort_name
})
finally:
# Assumes that the ecommerce service uses an API key to authenticate.
if has_api_key_permissions:
current_enrollment = api.get_enrollment(username, text_type(course_id))
audit_log(
'enrollment_change_requested',
course_id=text_type(course_id),
requested_mode=mode,
actual_mode=current_enrollment['mode'] if current_enrollment else None,
requested_activation=is_active,
actual_activation=current_enrollment['is_active'] if current_enrollment else None,
user_id=user.id
)
@can_disable_rate_limit
class CourseEnrollmentsApiListView(DeveloperErrorViewMixin, ListAPIView):
"""
**Use Cases**
Get a list of all course enrollments, optionally filtered by a course ID or list of usernames.
**Example Requests**
GET /api/enrollment/v1/enrollments
GET /api/enrollment/v1/enrollments?course_id={course_id}
GET /api/enrollment/v1/enrollments?username={username},{username},{username}
GET /api/enrollment/v1/enrollments?course_id={course_id}&username={username}
**Query Parameters for GET**
* course_id: Filters the result to course enrollments for the course corresponding to the
given course ID. The value must be URL encoded. Optional.
* username: List of comma-separated usernames. Filters the result to the course enrollments
of the given users. Optional.
* page_size: Number of results to return per page. Optional.
* page: Page number to retrieve. Optional.
**Response Values**
If the request for information about the course enrollments is successful, an HTTP 200 "OK" response
is returned.
The HTTP 200 response has the following values.
* results: A list of the course enrollments matching the request.
* created: Date and time when the course enrollment was created.
* mode: Mode for the course enrollment.
* is_active: Whether the course enrollment is active or not.
* user: Username of the user in the course enrollment.
* course_id: Course ID of the course in the course enrollment.
* next: The URL to the next page of results, or null if this is the
last page.
* previous: The URL to the next page of results, or null if this
is the first page.
If the user is not logged in, a 401 error is returned.
If the user is not global staff, a 403 error is returned.
If the specified course_id is not valid or any of the specified usernames
are not valid, a 400 error is returned.
If the specified course_id does not correspond to a valid course or if all the specified
usernames do not correspond to valid users, an HTTP 200 "OK" response is returned with an
empty 'results' field.
"""
authentication_classes = (
JwtAuthentication,
OAuth2AuthenticationAllowInactiveUser,
SessionAuthenticationAllowInactiveUser,
)
permission_classes = (permissions.IsAdminUser,)
throttle_classes = (EnrollmentUserThrottle,)
serializer_class = CourseEnrollmentsApiListSerializer
pagination_class = CourseEnrollmentsApiListPagination
def get_queryset(self):
"""
Get all the course enrollments for the given course_id and/or given list of usernames.
"""
form = CourseEnrollmentsApiListForm(self.request.query_params)
if not form.is_valid():
raise ValidationError(form.errors)
queryset = CourseEnrollment.objects.all()
course_id = form.cleaned_data.get('course_id')
usernames = form.cleaned_data.get('username')
if course_id:
queryset = queryset.filter(course_id=course_id)
if usernames:
queryset = queryset.filter(user__username__in=usernames)
return queryset

View File

@@ -10,7 +10,7 @@ import pytz
from django.test import TestCase
from social_django.models import UserSocialAuth
from enrollment import api
from openedx.core.djangoapps.enrollments import api
from openedx.core.djangoapps.user_api.models import RetirementState, UserRetirementStatus
from student.models import get_retired_email_by_email, get_retired_username_by_username
from student.tests.factories import UserFactory