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

Create Python API for program_enrollments: Part I (#21511)

Browse Source 
This is the first in a series of commits to create
a Python API for the LMS program_enrollments app.
We do some general refactoring, renaming, and clean-up
in order to move toward the creation of that API.

EDUCATOR-4321
This commit is contained in:
Kyle McCormick
2019-09-04 14:06:00 -04:00
committed by GitHub
parent b229e9749b
commit ea5652010c
24 changed files with 737 additions and 714 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

@@ -1,133 +0,0 @@
# -*- coding: utf-8 -*-
"""
ProgramEnrollment internal api
"""
from __future__ import absolute_import, unicode_literals
from datetime import datetime, timedelta
from pytz import UTC
from django.urls import reverse
from six import iteritems
from bulk_email.api import is_bulk_email_feature_enabled, is_user_opted_out_for_course
from edx_when.api import get_dates_for_course
from xmodule.modulestore.django import modulestore
from lms.djangoapps.program_enrollments.api.v1.constants import (
CourseRunProgressStatuses,
)
def get_due_dates(request, course_key, user):
"""
Get due date information for a user for blocks in a course.
Arguments:
request: the request object
course_key (CourseKey): the CourseKey for the course
user: the user object for which we want due date information
Returns:
due_dates (list): a list of dictionaries containing due date information
keys:
name: the display name of the block
url: the deep link to the block
date: the due date for the block
"""
dates = get_dates_for_course(
course_key,
user,
)
store = modulestore()
due_dates = []
for (block_key, date_type), date in iteritems(dates):
if date_type == 'due':
block = store.get_item(block_key)
# get url to the block in the course
block_url = reverse('jump_to', args=[course_key, block_key])
block_url = request.build_absolute_uri(block_url)
due_dates.append({
'name': block.display_name,
'url': block_url,
'date': date,
})
return due_dates
def get_course_run_url(request, course_id):
"""
Get the URL to a course run.
Arguments:
request: the request object
course_id (string): the course id of the course
Returns:
(string): the URL to the course run associated with course_id
"""
course_run_url = reverse('openedx.course_experience.course_home', args=[course_id])
return request.build_absolute_uri(course_run_url)
def get_emails_enabled(user, course_id):
"""
Get whether or not emails are enabled in the context of a course.
Arguments:
user: the user object for which we want to check whether emails are enabled
course_id (string): the course id of the course
Returns:
(bool): True if emails are enabled for the course associated with course_id for the user;
False otherwise
"""
if is_bulk_email_feature_enabled(course_id=course_id):
return not is_user_opted_out_for_course(user=user, course_id=course_id)
return None
def get_course_run_status(course_overview, certificate_info):
"""
Get the progress status of a course run, given the state of a user's certificate in the course.
In the case of self-paced course runs, the run is considered completed when either the course run has ended
OR the user has earned a passing certificate 30 days ago or longer.
Arguments:
course_overview (CourseOverview): the overview for the course run
certificate_info: A dict containing the following keys:
``is_passing``: whether the user has a passing certificate in the course run
``created``: the date the certificate was created
Returns:
status: one of (
CourseRunProgressStatuses.COMPLETE,
CourseRunProgressStatuses.IN_PROGRESS,
CourseRunProgressStatuses.UPCOMING,
)
"""
is_certificate_passing = certificate_info.get('is_passing', False)
certificate_creation_date = certificate_info.get('created', datetime.max)
if course_overview.pacing == 'instructor':
if course_overview.has_ended():
return CourseRunProgressStatuses.COMPLETED
elif course_overview.has_started():
return CourseRunProgressStatuses.IN_PROGRESS
else:
return CourseRunProgressStatuses.UPCOMING
elif course_overview.pacing == 'self':
thirty_days_ago = datetime.now(UTC) - timedelta(30)
certificate_completed = is_certificate_passing and (certificate_creation_date <= thirty_days_ago)
if course_overview.has_ended() or certificate_completed:
return CourseRunProgressStatuses.COMPLETED
elif course_overview.has_started():
return CourseRunProgressStatuses.IN_PROGRESS
else:
return CourseRunProgressStatuses.UPCOMING
return None

13
lms/djangoapps/program_enrollments/api/urls.py
View File

@@ -1,13 +0,0 @@
"""
Program Enrollment API URLs.
"""
from __future__ import absolute_import
from django.conf.urls import include, url
app_name = 'lms.djangoapps.program_enrollments'
urlpatterns = [
url(r'^v1/', include('program_enrollments.api.v1.urls', namespace='v1'))
]

0
lms/djangoapps/program_enrollments/api/v1/__init__.py
View File

65
lms/djangoapps/program_enrollments/api/v1/constants.py
View File

@@ -1,65 +0,0 @@
"""
Constants and strings for the course-enrollment app
"""
# Captures strings composed of alphanumeric characters a-f and dashes.
PROGRAM_UUID_PATTERN = r'(?P<program_uuid>[A-Fa-f0-9-]+)'
MAX_ENROLLMENT_RECORDS = 25
# The name of the key that identifies students for POST/PATCH requests
REQUEST_STUDENT_KEY = 'student_key'
ENABLE_ENROLLMENT_RESET_FLAG = 'ENABLE_ENROLLMENT_RESET'
class BaseEnrollmentResponseStatuses(object):
"""
Class to group common response statuses
"""
DUPLICATED = 'duplicated'
INVALID_STATUS = "invalid-status"
CONFLICT = "conflict"
ILLEGAL_OPERATION = "illegal-operation"
NOT_IN_PROGRAM = "not-in-program"
INTERNAL_ERROR = "internal-error"
ERROR_STATUSES = {
DUPLICATED,
INVALID_STATUS,
CONFLICT,
ILLEGAL_OPERATION,
NOT_IN_PROGRAM,
INTERNAL_ERROR,
}
class CourseEnrollmentResponseStatuses(BaseEnrollmentResponseStatuses):
"""
Class to group response statuses returned by the course enrollment endpoint
"""
ACTIVE = "active"
INACTIVE = "inactive"
NOT_FOUND = "not-found"
ERROR_STATUSES = BaseEnrollmentResponseStatuses.ERROR_STATUSES | {NOT_FOUND}
class ProgramEnrollmentResponseStatuses(BaseEnrollmentResponseStatuses):
"""
Class to group response statuses returned by the program enrollment endpoint
"""
ENROLLED = 'enrolled'
PENDING = 'pending'
SUSPENDED = 'suspended'
CANCELED = 'canceled'
VALID_STATUSES = [ENROLLED, PENDING, SUSPENDED, CANCELED]
class CourseRunProgressStatuses(object):
"""
Class to group statuses that a course run can be in with respect to user progress.
"""
IN_PROGRESS = 'in_progress'
UPCOMING = 'upcoming'
COMPLETED = 'completed'

227
lms/djangoapps/program_enrollments/api/v1/serializers.py
View File

@@ -1,227 +0,0 @@
"""
API Serializers
"""
from __future__ import absolute_import
from rest_framework import serializers
from six import text_type
from lms.djangoapps.program_enrollments.models import ProgramCourseEnrollment, ProgramEnrollment
from lms.djangoapps.program_enrollments.api.v1.constants import (
CourseRunProgressStatuses,
ProgramEnrollmentResponseStatuses
)
class InvalidStatusMixin(object):
"""
Mixin to provide has_invalid_status method
"""
def has_invalid_status(self):
"""
Returns whether or not this serializer has an invalid error choice on the "status" field
"""
try:
for status_error in self.errors['status']:
if status_error.code == 'invalid_choice':
return True
except KeyError:
pass
return False
# pylint: disable=abstract-method
class ProgramEnrollmentSerializer(serializers.ModelSerializer, InvalidStatusMixin):
"""
Serializer for Program Enrollments
"""
class Meta(object):
model = ProgramEnrollment
fields = ('user', 'external_user_key', 'program_uuid', 'curriculum_uuid', 'status')
validators = []
def validate(self, attrs):
""" This modifies self.instance in the case of updates """
if not self.instance:
enrollment = ProgramEnrollment(**attrs)
enrollment.full_clean()
else:
for key, value in attrs.items():
setattr(self.instance, key, value)
self.instance.full_clean()
return attrs
def create(self, validated_data):
return ProgramEnrollment.objects.create(**validated_data)
class BaseProgramEnrollmentRequestMixin(serializers.Serializer, InvalidStatusMixin):
"""
Base fields for all program enrollment related serializers
"""
student_key = serializers.CharField()
status = serializers.ChoiceField(
allow_blank=False,
choices=ProgramEnrollmentResponseStatuses.VALID_STATUSES
)
class ProgramEnrollmentCreateRequestSerializer(BaseProgramEnrollmentRequestMixin):
"""
Serializer for program enrollment creation requests
"""
curriculum_uuid = serializers.UUIDField()
class ProgramEnrollmentModifyRequestSerializer(BaseProgramEnrollmentRequestMixin):
"""
Serializer for program enrollment modification requests
"""
pass
class ProgramEnrollmentListSerializer(serializers.Serializer):
"""
Serializer for listing enrollments in a program.
"""
student_key = serializers.CharField(source='external_user_key')
status = serializers.CharField()
account_exists = serializers.SerializerMethodField()
curriculum_uuid = serializers.UUIDField()
class Meta(object):
model = ProgramEnrollment
def get_account_exists(self, obj):
return bool(obj.user)
# pylint: disable=abstract-method
class ProgramCourseEnrollmentRequestSerializer(serializers.Serializer, InvalidStatusMixin):
"""
Serializer for request to create a ProgramCourseEnrollment
"""
STATUS_CHOICES = ['active', 'inactive']
student_key = serializers.CharField(allow_blank=False)
status = serializers.ChoiceField(allow_blank=False, choices=STATUS_CHOICES)
class ProgramCourseEnrollmentListSerializer(serializers.Serializer):
"""
Serializer for listing course enrollments in a program.
"""
student_key = serializers.SerializerMethodField()
status = serializers.CharField()
account_exists = serializers.SerializerMethodField()
curriculum_uuid = serializers.SerializerMethodField()
class Meta(object):
model = ProgramCourseEnrollment
def get_student_key(self, obj):
return obj.program_enrollment.external_user_key
def get_account_exists(self, obj):
return bool(obj.program_enrollment.user)
def get_curriculum_uuid(self, obj):
return text_type(obj.program_enrollment.curriculum_uuid)
class ProgramCourseGradeResult(object):
"""
Represents a courserun grade for a user enrolled through a program.
Can be passed to ProgramCourseGradeResultSerializer.
"""
is_error = False
def __init__(self, program_course_enrollment, course_grade):
"""
Creates a new grade result given a ProgramCourseEnrollment object
and a course grade object.
"""
self.student_key = program_course_enrollment.program_enrollment.external_user_key
self.passed = course_grade.passed
self.percent = course_grade.percent
self.letter_grade = course_grade.letter_grade
class ProgramCourseGradeErrorResult(object):
"""
Represents a failure to load a courserun grade for a user enrolled through
a program.
Can be passed to ProgramCourseGradeResultSerializer.
"""
is_error = True
def __init__(self, program_course_enrollment, exception=None):
"""
Creates a new course grade error object given a
ProgramCourseEnrollment and an exception.
"""
self.student_key = program_course_enrollment.program_enrollment.external_user_key
self.error = text_type(exception) if exception else u"Unknown error"
class ProgramCourseGradeResultSerializer(serializers.Serializer):
"""
Serializer for a user's grade in a program courserun.
Meant to be used with ProgramCourseGradeResult
or ProgramCourseGradeErrorResult as input.
Absence of fields other than `student_key` will be ignored.
"""
# Required
student_key = serializers.CharField()
# From ProgramCourseGradeResult only
passed = serializers.BooleanField(required=False)
percent = serializers.FloatField(required=False)
letter_grade = serializers.CharField(required=False)
# From ProgramCourseGradeErrorResult only
error = serializers.CharField(required=False)
class DueDateSerializer(serializers.Serializer):
"""
Serializer for a due date.
"""
name = serializers.CharField()
url = serializers.CharField()
date = serializers.DateTimeField()
class CourseRunOverviewSerializer(serializers.Serializer):
"""
Serializer for a course run overview.
"""
STATUS_CHOICES = [
CourseRunProgressStatuses.IN_PROGRESS,
CourseRunProgressStatuses.UPCOMING,
CourseRunProgressStatuses.COMPLETED
]
course_run_id = serializers.CharField()
display_name = serializers.CharField()
resume_course_run_url = serializers.CharField(required=False)
course_run_url = serializers.CharField()
start_date = serializers.DateTimeField()
end_date = serializers.DateTimeField()
course_run_status = serializers.ChoiceField(allow_blank=False, choices=STATUS_CHOICES)
emails_enabled = serializers.BooleanField(required=False)
due_dates = serializers.ListField(child=DueDateSerializer())
micromasters_title = serializers.CharField(required=False)
certificate_download_url = serializers.CharField(required=False)
class CourseRunOverviewListSerializer(serializers.Serializer):
"""
Serializer for a list of course run overviews.
"""
course_runs = serializers.ListField(child=CourseRunOverviewSerializer())

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

46
lms/djangoapps/program_enrollments/api/v1/tests/test_serializers.py
View File

@@ -1,46 +0,0 @@
"""
Unit tests for ProgramEnrollment serializers.
"""
from __future__ import absolute_import, unicode_literals
from uuid import uuid4
from django.test import TestCase
from lms.djangoapps.program_enrollments.api.v1.serializers import ProgramEnrollmentSerializer
from lms.djangoapps.program_enrollments.models import ProgramEnrollment
from student.tests.factories import UserFactory
class ProgramEnrollmentSerializerTests(TestCase):
"""
Tests for the ProgramEnrollment serializer.
"""
def setUp(self):
"""
Set up the test data used in the specific tests
"""
super(ProgramEnrollmentSerializerTests, self).setUp()
self.user = UserFactory.create()
self.enrollment = ProgramEnrollment.objects.create(
user=self.user,
external_user_key='abc',
program_uuid=uuid4(),
curriculum_uuid=uuid4(),
status='enrolled'
)
self.serializer = ProgramEnrollmentSerializer(instance=self.enrollment)
def test_serializer_contains_expected_fields(self):
data = self.serializer.data
self.assertEqual(
set(data.keys()),
set([
'user',
'external_user_key',
'program_uuid',
'curriculum_uuid',
'status'
])
)

2116
lms/djangoapps/program_enrollments/api/v1/tests/test_views.py
View File

@@ -1,2116 +0,0 @@
"""
Unit tests for ProgramEnrollment views.
"""
from __future__ import absolute_import, unicode_literals
import json
from datetime import datetime, timedelta
from uuid import UUID, uuid4
import ddt
import mock
from django.conf import settings
from django.contrib.auth.models import User
from django.core.cache import cache
from django.test import override_settings
from django.urls import reverse
from freezegun import freeze_time
from opaque_keys.edx.keys import CourseKey
from organizations.tests.factories import OrganizationFactory
from pytz import UTC
from rest_framework import status
from rest_framework.test import APITestCase
from six import text_type
from six.moves import range, zip
from bulk_email.models import BulkEmailFlag, Optout
from course_modes.models import CourseMode
from lms.djangoapps.certificates.models import CertificateStatuses
from lms.djangoapps.certificates.tests.factories import GeneratedCertificateFactory
from lms.djangoapps.courseware.tests.factories import GlobalStaffFactory, InstructorFactory
from lms.djangoapps.program_enrollments.api.v1.constants import (
ENABLE_ENROLLMENT_RESET_FLAG,
MAX_ENROLLMENT_RECORDS,
REQUEST_STUDENT_KEY
)
from lms.djangoapps.program_enrollments.api.v1.constants import CourseEnrollmentResponseStatuses as CourseStatuses
from lms.djangoapps.program_enrollments.api.v1.constants import CourseRunProgressStatuses
from lms.djangoapps.program_enrollments.api.v1.constants import ProgramEnrollmentResponseStatuses as ProgramStatuses
from lms.djangoapps.program_enrollments.models import ProgramCourseEnrollment, ProgramEnrollment
from lms.djangoapps.program_enrollments.tests.factories import ProgramCourseEnrollmentFactory, ProgramEnrollmentFactory
from lms.djangoapps.program_enrollments.utils import ProviderDoesNotExistException
from openedx.core.djangoapps.catalog.cache import PROGRAM_CACHE_KEY_TPL, PROGRAMS_BY_ORGANIZATION_CACHE_KEY_TPL
from openedx.core.djangoapps.catalog.tests.factories import CourseFactory, CourseRunFactory
from openedx.core.djangoapps.catalog.tests.factories import OrganizationFactory as CatalogOrganizationFactory
from openedx.core.djangoapps.catalog.tests.factories import ProgramFactory
from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
from openedx.core.djangolib.testing.utils import CacheIsolationMixin
from student.roles import CourseStaffRole
from student.tests.factories import CourseEnrollmentFactory, UserFactory
from third_party_auth.tests.factories import SAMLProviderConfigFactory
from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory as ModulestoreCourseFactory
from xmodule.modulestore.tests.factories import ItemFactory
class ProgramCacheTestCaseMixin(CacheIsolationMixin):
"""
Mixin for using program cache in tests
"""
ENABLED_CACHES = ['default']
@staticmethod
def setup_catalog_cache(program_uuid, organization_key):
"""
helper function to initialize a cached program with an single authoring_organization
"""
catalog_org = CatalogOrganizationFactory.create(key=organization_key)
program = ProgramFactory.create(
uuid=program_uuid,
authoring_organizations=[catalog_org]
)
cache.set(PROGRAM_CACHE_KEY_TPL.format(uuid=program_uuid), program, None)
return program
@staticmethod
def set_program_in_catalog_cache(program_uuid, program):
cache.set(PROGRAM_CACHE_KEY_TPL.format(uuid=program_uuid), program, None)
@staticmethod
def set_org_in_catalog_cache(organization, program_uuids):
cache.set(PROGRAMS_BY_ORGANIZATION_CACHE_KEY_TPL.format(org_key=organization.short_name), program_uuids)
class ListViewTestMixin(ProgramCacheTestCaseMixin):
"""
Mixin to define some shared test data objects for program/course enrollment
list view tests.
"""
view_name = None
@classmethod
def setUpClass(cls):
super(ListViewTestMixin, cls).setUpClass()
cls.start_cache_isolation()
cls.program_uuid = '00000000-1111-2222-3333-444444444444'
cls.program_uuid_tmpl = '00000000-1111-2222-3333-4444444444{0:02d}'
cls.curriculum_uuid = 'aaaaaaaa-1111-2222-3333-444444444444'
cls.other_curriculum_uuid = 'bbbbbbbb-1111-2222-3333-444444444444'
cls.organization_key = "orgkey"
cls.program = cls.setup_catalog_cache(cls.program_uuid, cls.organization_key)
cls.course_id = CourseKey.from_string('course-v1:edX+ToyX+Toy_Course')
_ = CourseOverviewFactory.create(id=cls.course_id)
cls.password = 'password'
cls.student = UserFactory.create(username='student', password=cls.password)
cls.global_staff = GlobalStaffFactory.create(username='global-staff', password=cls.password)
@classmethod
def tearDownClass(cls):
super(ListViewTestMixin, cls).tearDownClass()
cls.end_cache_isolation()
def setUp(self):
super(ListViewTestMixin, self).setUp()
self.set_program_in_catalog_cache(self.program_uuid, self.program)
self.curriculum = next(c for c in self.program['curricula'] if c['is_active'])
self.course = self.curriculum['courses'][0]
self.course_run = self.course["course_runs"][0]
self.course_key = CourseKey.from_string(self.course_run["key"])
CourseOverviewFactory(id=self.course_key)
self.course_not_in_program = CourseFactory()
self.course_not_in_program_key = CourseKey.from_string(
self.course_not_in_program["course_runs"][0]["key"]
)
CourseOverviewFactory(id=self.course_not_in_program_key)
def get_url(self, program_uuid=None, course_id=None):
""" Returns the primary URL requested by the test case. """
kwargs = {'program_uuid': program_uuid or self.program_uuid}
if course_id:
kwargs['course_id'] = course_id
return reverse(self.view_name, kwargs=kwargs)
def log_in_non_staff(self):
self.client.login(username=self.student.username, password=self.password)
def log_in_staff(self):
self.client.login(username=self.global_staff.username, password=self.password)
@ddt.ddt
class UserProgramReadOnlyAccessViewTest(ListViewTestMixin, APITestCase):
"""
Tests for the UserProgramReadonlyAccess view class
"""
view_name = 'programs_api:v1:user_program_readonly_access'
@classmethod
def setUpClass(cls):
super(UserProgramReadOnlyAccessViewTest, cls).setUpClass()
cls.mock_program_data = [
{'uuid': cls.program_uuid_tmpl.format(11), 'marketing_slug': 'garbage-program', 'type': 'masters'},
{'uuid': cls.program_uuid_tmpl.format(22), 'marketing_slug': 'garbage-study', 'type': 'micromaster'},
{'uuid': cls.program_uuid_tmpl.format(33), 'marketing_slug': 'garbage-life', 'type': 'masters'},
]
cls.course_staff = InstructorFactory.create(password=cls.password, course_key=cls.course_id)
cls.date = datetime(2013, 1, 22, tzinfo=UTC)
CourseEnrollmentFactory(
course_id=cls.course_id,
user=cls.course_staff,
created=cls.date,
)
def test_401_if_anonymous(self):
response = self.client.get(reverse(self.view_name))
assert status.HTTP_401_UNAUTHORIZED == response.status_code
@ddt.data(
('masters', 2),
('micromaster', 1)
)
@ddt.unpack
def test_global_staff(self, program_type, expected_data_size):
self.client.login(username=self.global_staff.username, password=self.password)
mock_return_value = [program for program in self.mock_program_data if program['type'] == program_type]
with mock.patch(
'lms.djangoapps.program_enrollments.api.v1.views.get_programs_by_type',
autospec=True,
return_value=mock_return_value
) as mock_get_programs_by_type:
response = self.client.get(reverse(self.view_name) + '?type=' + program_type)
assert status.HTTP_200_OK == response.status_code
assert len(response.data) == expected_data_size
mock_get_programs_by_type.assert_called_once_with(response.wsgi_request.site, program_type)
def test_course_staff(self):
self.client.login(username=self.course_staff.username, password=self.password)
with mock.patch(
'lms.djangoapps.program_enrollments.api.v1.views.get_programs',
autospec=True,
return_value=[self.mock_program_data[0]]
) as mock_get_programs:
response = self.client.get(reverse(self.view_name) + '?type=masters')
assert status.HTTP_200_OK == response.status_code
assert len(response.data) == 1
mock_get_programs.assert_called_once_with(course=self.course_id)
def test_course_staff_of_multiple_courses(self):
other_course_key = CourseKey.from_string('course-v1:edX+ToyX+Other_Course')
CourseEnrollmentFactory.create(course_id=other_course_key, user=self.course_staff)
CourseStaffRole(other_course_key).add_users(self.course_staff)
self.client.login(username=self.course_staff.username, password=self.password)
with mock.patch(
'lms.djangoapps.program_enrollments.api.v1.views.get_programs',
autospec=True,
side_effect=[[self.mock_program_data[0]], [self.mock_program_data[2]]]
) as mock_get_programs:
response = self.client.get(reverse(self.view_name) + '?type=masters')
assert status.HTTP_200_OK == response.status_code
assert len(response.data) == 2
mock_get_programs.assert_has_calls([
mock.call(course=self.course_id),
mock.call(course=other_course_key),
], any_order=True)
@mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True, return_value=None)
def test_learner_200_if_no_programs_enrolled(self, mock_get_programs):
self.client.login(username=self.student.username, password=self.password)
response = self.client.get(reverse(self.view_name))
assert status.HTTP_200_OK == response.status_code
assert response.data == []
mock_get_programs.assert_called_once_with(uuids=[])
def test_learner_200_many_programs(self):
for program in self.mock_program_data:
ProgramEnrollmentFactory.create(
program_uuid=program['uuid'],
curriculum_uuid=self.curriculum_uuid,
user=self.student,
status='pending',
external_user_key='user-{}'.format(self.student.id),
)
self.client.login(username=self.student.username, password=self.password)
with mock.patch(
'lms.djangoapps.program_enrollments.api.v1.views.get_programs',
autospec=True,
return_value=self.mock_program_data
) as mock_get_programs:
response = self.client.get(reverse(self.view_name))
assert status.HTTP_200_OK == response.status_code
assert len(response.data) == 3
mock_get_programs.assert_called_once_with(uuids=[UUID(item['uuid']) for item in self.mock_program_data])
class ProgramEnrollmentListTest(ListViewTestMixin, APITestCase):
"""
Tests for GET calls to the Program Enrollments API.
"""
view_name = 'programs_api:v1:program_enrollments'
def create_program_enrollments(self):
"""
Helper method for creating program enrollment records.
"""
for i in range(2):
user_key = 'user-{}'.format(i)
ProgramEnrollmentFactory.create(
program_uuid=self.program_uuid,
curriculum_uuid=self.curriculum_uuid,
user=None,
status='pending',
external_user_key=user_key,
)
for i in range(2, 4):
user_key = 'user-{}'.format(i)
ProgramEnrollmentFactory.create(
program_uuid=self.program_uuid, curriculum_uuid=self.curriculum_uuid, external_user_key=user_key,
)
self.addCleanup(self.destroy_program_enrollments)
def destroy_program_enrollments(self):
"""
Deletes program enrollments associated with this test case's program_uuid.
"""
ProgramEnrollment.objects.filter(program_uuid=self.program_uuid).delete()
@mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True, return_value=None)
def test_404_if_no_program_with_key(self, mock_get_programs):
self.client.login(username=self.global_staff.username, password=self.password)
response = self.client.get(self.get_url(self.program_uuid))
assert status.HTTP_404_NOT_FOUND == response.status_code
mock_get_programs.assert_called_once_with(uuid=self.program_uuid)
def test_403_if_not_staff(self):
self.client.login(username=self.student.username, password=self.password)
response = self.client.get(self.get_url(self.program_uuid))
assert status.HTTP_403_FORBIDDEN == response.status_code
def test_401_if_anonymous(self):
response = self.client.get(self.get_url(self.program_uuid))
assert status.HTTP_401_UNAUTHORIZED == response.status_code
def test_200_empty_results(self):
self.client.login(username=self.global_staff.username, password=self.password)
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
response = self.client.get(self.get_url(self.program_uuid))
assert status.HTTP_200_OK == response.status_code
expected = {
'next': None,
'previous': None,
'results': [],
}
assert expected == response.data
def test_200_many_results(self):
self.client.login(username=self.global_staff.username, password=self.password)
self.create_program_enrollments()
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
response = self.client.get(self.get_url(self.program_uuid))
assert status.HTTP_200_OK == response.status_code
expected = {
'next': None,
'previous': None,
'results': [
{
'student_key': 'user-0', 'status': 'pending', 'account_exists': False,
'curriculum_uuid': text_type(self.curriculum_uuid),
},
{
'student_key': 'user-1', 'status': 'pending', 'account_exists': False,
'curriculum_uuid': text_type(self.curriculum_uuid),
},
{
'student_key': 'user-2', 'status': 'enrolled', 'account_exists': True,
'curriculum_uuid': text_type(self.curriculum_uuid),
},
{
'student_key': 'user-3', 'status': 'enrolled', 'account_exists': True,
'curriculum_uuid': text_type(self.curriculum_uuid),
},
],
}
assert expected == response.data
def test_200_many_pages(self):
self.client.login(username=self.global_staff.username, password=self.password)
self.create_program_enrollments()
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
url = self.get_url(self.program_uuid) + '?page_size=2'
response = self.client.get(url)
assert status.HTTP_200_OK == response.status_code
expected_results = [
{
'student_key': 'user-0', 'status': 'pending', 'account_exists': False,
'curriculum_uuid': text_type(self.curriculum_uuid),
},
{
'student_key': 'user-1', 'status': 'pending', 'account_exists': False,
'curriculum_uuid': text_type(self.curriculum_uuid),
},
]
assert expected_results == response.data['results']
# there's going to be a 'cursor' query param, but we have no way of knowing it's value
assert response.data['next'] is not None
assert self.get_url(self.program_uuid) in response.data['next']
assert '?cursor=' in response.data['next']
assert response.data['previous'] is None
next_response = self.client.get(response.data['next'])
assert status.HTTP_200_OK == next_response.status_code
next_expected_results = [
{
'student_key': 'user-2', 'status': 'enrolled', 'account_exists': True,
'curriculum_uuid': text_type(self.curriculum_uuid),
},
{
'student_key': 'user-3', 'status': 'enrolled', 'account_exists': True,
'curriculum_uuid': text_type(self.curriculum_uuid),
},
]
assert next_expected_results == next_response.data['results']
assert next_response.data['next'] is None
# there's going to be a 'cursor' query param, but we have no way of knowing it's value
assert next_response.data['previous'] is not None
assert self.get_url(self.program_uuid) in next_response.data['previous']
assert '?cursor=' in next_response.data['previous']
class ProgramEnrollmentDataMixin(object):
""" Provides methods for creating ProgramEnrollments and ProgramCourseEnrollments. """
def learner_enrollment(self, student_key, enrollment_status="active"):
"""
Convenience method to create a learner enrollment record
"""
return {"student_key": student_key, "status": enrollment_status}
def request(self, path, data):
pass
def prepare_student(self, key):
pass
def create_program_enrollment(self, external_user_key, user=False):
"""
Creates and returns a ProgramEnrollment for the given external_user_key and
user if specified.
"""
program_enrollment = ProgramEnrollmentFactory.create(
external_user_key=external_user_key,
program_uuid=self.program_uuid,
)
if user is not False:
program_enrollment.user = user
program_enrollment.save()
return program_enrollment
def create_program_course_enrollment(self, program_enrollment, course_status='active'):
"""
Creates and returns a ProgramCourseEnrollment for the given program_enrollment and
self.course_key, creating a CourseEnrollment if the program enrollment has a user
"""
course_enrollment = None
if program_enrollment.user:
course_enrollment = CourseEnrollmentFactory.create(
course_id=self.course_key,
user=program_enrollment.user,
mode=CourseMode.MASTERS
)
course_enrollment.is_active = course_status == "active"
course_enrollment.save()
return ProgramCourseEnrollmentFactory.create(
program_enrollment=program_enrollment,
course_key=self.course_key,
course_enrollment=course_enrollment,
status=course_status,
)
def create_program_and_course_enrollments(self, external_user_key, user=False, course_status='active'):
program_enrollment = self.create_program_enrollment(external_user_key, user)
return self.create_program_course_enrollment(program_enrollment, course_status=course_status)
@ddt.ddt
class BaseCourseEnrollmentTestsMixin(ProgramEnrollmentDataMixin, ListViewTestMixin, ProgramCacheTestCaseMixin):
"""
A base for tests for course enrollment.
Children should override self.request()
"""
view_name = 'programs_api:v1:program_course_enrollments'
@classmethod
def setUpClass(cls):
super(BaseCourseEnrollmentTestsMixin, cls).setUpClass()
cls.start_cache_isolation()
@classmethod
def tearDownClass(cls):
cls.end_cache_isolation()
super(BaseCourseEnrollmentTestsMixin, cls).tearDownClass()
def setUp(self):
super(BaseCourseEnrollmentTestsMixin, self).setUp()
self.default_url = self.get_url(self.program_uuid, self.course_key)
self.log_in_staff()
def assert_program_course_enrollment(self, external_user_key, expected_status, has_user, mode=CourseMode.MASTERS):
"""
Convenience method to assert that a ProgramCourseEnrollment exists,
and potentially that a CourseEnrollment also exists
"""
enrollment = ProgramCourseEnrollment.objects.get(
program_enrollment__external_user_key=external_user_key,
program_enrollment__program_uuid=self.program_uuid
)
self.assertEqual(expected_status, enrollment.status)
self.assertEqual(self.course_key, enrollment.course_key)
course_enrollment = enrollment.course_enrollment
if has_user:
self.assertIsNotNone(course_enrollment)
self.assertEqual(expected_status == "active", course_enrollment.is_active)
self.assertEqual(self.course_key, course_enrollment.course_id)
self.assertEqual(mode, course_enrollment.mode)
else:
self.assertIsNone(course_enrollment)
def test_401_not_logged_in(self):
self.client.logout()
request_data = [self.learner_enrollment("learner-1")]
response = self.request(self.default_url, request_data)
self.assertEqual(401, response.status_code)
def test_403_forbidden(self):
self.client.logout()
self.log_in_non_staff()
request_data = [self.learner_enrollment("learner-1")]
response = self.request(self.default_url, request_data)
self.assertEqual(403, response.status_code)
def test_413_payload_too_large(self):
request_data = [self.learner_enrollment(str(i)) for i in range(30)]
response = self.request(self.default_url, request_data)
self.assertEqual(413, response.status_code)
def test_404_not_found(self):
nonexistant_course_key = CourseKey.from_string("course-v1:fake+fake+fake")
paths = [
self.get_url(uuid4(), self.course_key), # program not found
self.get_url(self.program_uuid, nonexistant_course_key), # course not found
self.get_url(self.program_uuid, self.course_not_in_program_key), # course not in program
]
request_data = [self.learner_enrollment("learner-1")]
for path_404 in paths:
response = self.request(path_404, request_data)
self.assertEqual(404, response.status_code)
def test_404_no_curriculum(self):
with mock.patch.dict(self.program, curricula=[]):
self.set_program_in_catalog_cache(self.program_uuid, self.program)
request_data = [self.learner_enrollment("learner-1")]
response = self.request(self.default_url, request_data)
self.assertEqual(404, response.status_code)
def test_duplicate_learner(self):
request_data = [
self.learner_enrollment("learner-1", "active"),
self.learner_enrollment("learner-1", "active"),
]
response = self.request(self.default_url, request_data)
self.assertEqual(422, response.status_code)
self.assertDictEqual(
{
"learner-1": CourseStatuses.DUPLICATED
},
response.data
)
def test_user_not_in_program(self):
request_data = [
self.learner_enrollment("learner-1"),
]
response = self.request(self.default_url, request_data)
self.assertEqual(422, response.status_code)
self.assertDictEqual(
{
"learner-1": CourseStatuses.NOT_IN_PROGRAM,
},
response.data
)
def test_invalid_status(self):
request_data = [self.learner_enrollment('learner-1', 'this-is-not-a-status')]
response = self.request(self.default_url, request_data)
self.assertEqual(422, response.status_code)
self.assertDictEqual({'learner-1': CourseStatuses.INVALID_STATUS}, response.data)
@ddt.data(
[{'status': 'active'}],
[{'student_key': '000'}],
["this isn't even a dict!"],
[{'student_key': '000', 'status': 'active'}, "bad_data"],
"not a list",
)
def test_422_unprocessable_entity_bad_data(self, request_data):
response = self.request(self.default_url, request_data)
self.assertEqual(response.status_code, 400)
self.assertIn('invalid enrollment record', response.data)
@ddt.data(
[{'status': 'pending'}],
[{'status': 'not-a-status'}],
[{'status': 'pending'}, {'status': 'pending'}],
)
def test_no_student_key(self, bad_records):
request_data = [self.learner_enrollment('learner-1')]
request_data.extend(bad_records)
response = self.request(self.default_url, request_data)
self.assertEqual(response.status_code, 400)
self.assertIn('invalid enrollment record', response.data)
def test_extra_field(self):
self.prepare_student('learner-1')
enrollment = self.learner_enrollment('learner-1', 'inactive')
enrollment['favorite_author'] = 'Hemingway'
request_data = [enrollment]
response = self.request(self.default_url, request_data)
self.assertEqual(response.status_code, 200)
self.assertDictEqual(
response.data,
{'learner-1': 'inactive'}
)
class CourseEnrollmentPostTests(BaseCourseEnrollmentTestsMixin, APITestCase):
""" Tests for course enrollment POST """
def request(self, path, data):
return self.client.post(path, data, format='json')
def prepare_student(self, key):
self.create_program_enrollment(key)
def test_create_enrollments(self):
self.create_program_enrollment('learner-1')
self.create_program_enrollment('learner-2')
self.create_program_enrollment('learner-3', user=None)
self.create_program_enrollment('learner-4', user=None)
post_data = [
self.learner_enrollment("learner-1", "active"),
self.learner_enrollment("learner-2", "inactive"),
self.learner_enrollment("learner-3", "active"),
self.learner_enrollment("learner-4", "inactive"),
]
response = self.request(self.default_url, post_data)
self.assertEqual(200, response.status_code)
self.assertDictEqual(
{
"learner-1": "active",
"learner-2": "inactive",
"learner-3": "active",
"learner-4": "inactive",
},
response.data
)
self.assert_program_course_enrollment("learner-1", "active", True)
self.assert_program_course_enrollment("learner-2", "inactive", True)
self.assert_program_course_enrollment("learner-3", "active", False)
self.assert_program_course_enrollment("learner-4", "inactive", False)
def test_program_course_enrollment_exists(self):
"""
The program enrollments application already has a program_course_enrollment
record for this user and course
"""
self.create_program_and_course_enrollments('learner-1')
post_data = [self.learner_enrollment("learner-1")]
response = self.request(self.default_url, post_data)
self.assertEqual(422, response.status_code)
self.assertDictEqual({'learner-1': CourseStatuses.CONFLICT}, response.data)
def test_user_currently_enrolled_in_course(self):
"""
If a user is already enrolled in a course through a different method
that enrollment should be linked but not overwritten as masters.
"""
CourseEnrollmentFactory.create(
course_id=self.course_key,
user=self.student,
mode=CourseMode.VERIFIED
)
self.create_program_enrollment('learner-1', user=self.student)
post_data = [
self.learner_enrollment("learner-1", "active")
]
response = self.request(self.default_url, post_data)
self.assertEqual(200, response.status_code)
self.assertDictEqual(
{
"learner-1": "active"
},
response.data
)
self.assert_program_course_enrollment("learner-1", "active", True, mode=CourseMode.VERIFIED)
def test_207_multistatus(self):
self.create_program_enrollment('learner-1')
post_data = [self.learner_enrollment("learner-1"), self.learner_enrollment("learner-2")]
response = self.request(self.default_url, post_data)
self.assertEqual(207, response.status_code)
self.assertDictEqual(
{'learner-1': CourseStatuses.ACTIVE, 'learner-2': CourseStatuses.NOT_IN_PROGRAM},
response.data
)
@ddt.ddt
class CourseEnrollmentModificationTestMixin(BaseCourseEnrollmentTestsMixin):
"""
Base class for both the PATCH and PUT endpoints for Course Enrollment API
Children needs to implement assert_user_not_enrolled_test_result and
setup_change_test_data
"""
def prepare_student(self, key):
self.create_program_and_course_enrollments(key)
def test_207_multistatus(self):
self.create_program_and_course_enrollments('learner-1')
mod_data = [self.learner_enrollment("learner-1"), self.learner_enrollment("learner-2")]
response = self.request(self.default_url, mod_data)
self.assertEqual(207, response.status_code)
self.assertDictEqual(
{'learner-1': CourseStatuses.ACTIVE, 'learner-2': CourseStatuses.NOT_IN_PROGRAM},
response.data
)
def test_user_not_enrolled_in_course(self):
self.create_program_enrollment('learner-1')
patch_data = [self.learner_enrollment('learner-1')]
response = self.request(self.default_url, patch_data)
self.assert_user_not_enrolled_test_result(response)
def assert_user_not_enrolled_test_result(self, response):
pass
def setup_change_test_data(self, initial_statuses):
pass
@ddt.data(
('active', 'inactive', 'active', 'inactive'),
('inactive', 'active', 'inactive', 'active'),
('active', 'active', 'active', 'active'),
('inactive', 'inactive', 'inactive', 'inactive'),
)
def test_change_status(self, initial_statuses):
self.setup_change_test_data(initial_statuses)
mod_data = [
self.learner_enrollment('learner-1', 'inactive'),
self.learner_enrollment('learner-2', 'active'),
self.learner_enrollment('learner-3', 'inactive'),
self.learner_enrollment('learner-4', 'active'),
]
response = self.request(self.default_url, mod_data)
self.assertEqual(200, response.status_code)
self.assertDictEqual(
{
'learner-1': 'inactive',
'learner-2': 'active',
'learner-3': 'inactive',
'learner-4': 'active',
},
response.data
)
self.assert_program_course_enrollment('learner-1', 'inactive', True)
self.assert_program_course_enrollment('learner-2', 'active', True)
self.assert_program_course_enrollment('learner-3', 'inactive', False)
self.assert_program_course_enrollment('learner-4', 'active', False)
class CourseEnrollmentPatchTests(CourseEnrollmentModificationTestMixin, APITestCase):
""" Tests for course enrollment PATCH """
def request(self, path, data):
return self.client.patch(path, data, format='json')
def assert_user_not_enrolled_test_result(self, response):
self.assertEqual(422, response.status_code)
self.assertDictEqual({'learner-1': CourseStatuses.NOT_FOUND}, response.data)
def setup_change_test_data(self, initial_statuses):
self.create_program_and_course_enrollments('learner-1', course_status=initial_statuses[0])
self.create_program_and_course_enrollments('learner-2', course_status=initial_statuses[1])
self.create_program_and_course_enrollments('learner-3', course_status=initial_statuses[2], user=None)
self.create_program_and_course_enrollments('learner-4', course_status=initial_statuses[3], user=None)
class CourseEnrollmentPutTests(CourseEnrollmentModificationTestMixin, APITestCase):
""" Tests for course enrollment PUT """
def request(self, path, data):
return self.client.put(path, data, format='json')
def assert_user_not_enrolled_test_result(self, response):
self.assertEqual(200, response.status_code)
self.assertDictEqual({'learner-1': CourseStatuses.ACTIVE}, response.data)
def setup_change_test_data(self, initial_statuses):
self.create_program_and_course_enrollments('learner-1', course_status=initial_statuses[0])
self.create_program_enrollment('learner-2')
self.create_program_enrollment('learner-3', user=None)
self.create_program_and_course_enrollments('learner-4', course_status=initial_statuses[3], user=None)
class ProgramCourseEnrollmentListTest(ListViewTestMixin, APITestCase):
"""
Tests for GET calls to the Program Course Enrollments API.
"""
view_name = 'programs_api:v1:program_course_enrollments'
def create_course_enrollments(self):
""" Helper method for creating ProgramCourseEnrollments. """
program_enrollment_1 = ProgramEnrollmentFactory.create(
program_uuid=self.program_uuid, curriculum_uuid=self.curriculum_uuid, external_user_key='user-0',
)
program_enrollment_2 = ProgramEnrollmentFactory.create(
program_uuid=self.program_uuid, curriculum_uuid=self.other_curriculum_uuid, external_user_key='user-0',
)
ProgramCourseEnrollmentFactory.create(
program_enrollment=program_enrollment_1,
course_key=self.course_id,
status='active',
)
ProgramCourseEnrollmentFactory.create(
program_enrollment=program_enrollment_2,
course_key=self.course_id,
status='inactive',
)
self.addCleanup(self.destroy_course_enrollments)
def destroy_course_enrollments(self):
""" Helper method for tearing down ProgramCourseEnrollments. """
ProgramCourseEnrollment.objects.filter(
program_enrollment__program_uuid=self.program_uuid,
course_key=self.course_id
).delete()
@mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True, return_value=None)
def test_404_if_no_program_with_key(self, mock_get_programs):
self.client.login(username=self.global_staff.username, password=self.password)
response = self.client.get(self.get_url(self.program_uuid, self.course_id))
assert status.HTTP_404_NOT_FOUND == response.status_code
mock_get_programs.assert_called_once_with(uuid=self.program_uuid)
def test_404_if_course_does_not_exist(self):
other_course_key = CourseKey.from_string('course-v1:edX+ToyX+Other_Course')
self.client.login(username=self.global_staff.username, password=self.password)
response = self.client.get(self.get_url(self.program_uuid, other_course_key))
assert status.HTTP_404_NOT_FOUND == response.status_code
def test_403_if_not_staff(self):
self.client.login(username=self.student.username, password=self.password)
response = self.client.get(self.get_url(self.program_uuid, self.course_id))
assert status.HTTP_403_FORBIDDEN == response.status_code
def test_401_if_anonymous(self):
response = self.client.get(self.get_url(self.program_uuid, self.course_id))
assert status.HTTP_401_UNAUTHORIZED == response.status_code
def test_200_empty_results(self):
self.client.login(username=self.global_staff.username, password=self.password)
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
response = self.client.get(self.get_url(self.program_uuid, self.course_id))
assert status.HTTP_200_OK == response.status_code
expected = {
'next': None,
'previous': None,
'results': [],
}
assert expected == response.data
def test_200_many_results(self):
self.client.login(username=self.global_staff.username, password=self.password)
self.create_course_enrollments()
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
response = self.client.get(self.get_url(self.program_uuid, self.course_id))
assert status.HTTP_200_OK == response.status_code
expected = {
'next': None,
'previous': None,
'results': [
{
'student_key': 'user-0', 'status': 'active', 'account_exists': True,
'curriculum_uuid': text_type(self.curriculum_uuid),
},
{
'student_key': 'user-0', 'status': 'inactive', 'account_exists': True,
'curriculum_uuid': text_type(self.other_curriculum_uuid),
},
],
}
assert expected == response.data
def test_200_many_pages(self):
self.client.login(username=self.global_staff.username, password=self.password)
self.create_course_enrollments()
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
url = self.get_url(self.program_uuid, self.course_id) + '?page_size=1'
response = self.client.get(url)
assert status.HTTP_200_OK == response.status_code
expected_results = [
{
'student_key': 'user-0', 'status': 'active', 'account_exists': True,
'curriculum_uuid': text_type(self.curriculum_uuid),
},
]
assert expected_results == response.data['results']
# there's going to be a 'cursor' query param, but we have no way of knowing it's value
assert response.data['next'] is not None
assert self.get_url(self.program_uuid, self.course_id) in response.data['next']
assert '?cursor=' in response.data['next']
assert response.data['previous'] is None
next_response = self.client.get(response.data['next'])
assert status.HTTP_200_OK == next_response.status_code
next_expected_results = [
{
'student_key': 'user-0', 'status': 'inactive', 'account_exists': True,
'curriculum_uuid': text_type(self.other_curriculum_uuid),
},
]
assert next_expected_results == next_response.data['results']
assert next_response.data['next'] is None
# there's going to be a 'cursor' query param, but we have no way of knowing it's value
assert next_response.data['previous'] is not None
assert self.get_url(self.program_uuid, self.course_id) in next_response.data['previous']
assert '?cursor=' in next_response.data['previous']
@ddt.ddt
class BaseProgramEnrollmentWriteTestsMixin(object):
""" Mixin class that defines common tests for program enrollment write endpoints """
add_uuid = False
program_uuid = '00000000-1111-2222-3333-444444444444'
success_status = 200
def student_enrollment(self, enrollment_status, external_user_key=None, prepare_student=False):
""" Convenience method to create a student enrollment record """
enrollment = {
REQUEST_STUDENT_KEY: external_user_key or str(uuid4().hex[0:10]),
'status': enrollment_status,
}
if self.add_uuid:
enrollment['curriculum_uuid'] = str(uuid4())
if prepare_student:
self.prepare_student(enrollment)
return enrollment
def prepare_student(self, enrollment):
pass
def get_url(self, program_uuid=None):
if program_uuid is None:
program_uuid = uuid4()
return reverse('programs_api:v1:program_enrollments', args=[program_uuid])
def test_unauthenticated(self):
self.client.logout()
request_data = [self.student_enrollment('enrolled')]
response = self.request(self.get_url(), json.dumps(request_data), content_type='application/json')
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
def test_enrollment_payload_limit(self):
request_data = [self.student_enrollment('enrolled') for _ in range(MAX_ENROLLMENT_RECORDS + 1)]
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
response = self.request(self.get_url(), json.dumps(request_data), content_type='application/json')
self.assertEqual(response.status_code, status.HTTP_413_REQUEST_ENTITY_TOO_LARGE)
def test_duplicate_enrollment(self):
request_data = [
self.student_enrollment('enrolled', '001'),
self.student_enrollment('enrolled', '001'),
]
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
response = self.request(self.get_url(), json.dumps(request_data), content_type='application/json')
self.assertEqual(response.status_code, status.HTTP_422_UNPROCESSABLE_ENTITY)
self.assertEqual(response.data, {'001': 'duplicated'})
def test_unprocessable_enrollment(self):
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
response = self.request(
self.get_url(),
json.dumps([{'status': 'enrolled'}]),
content_type='application/json'
)
self.assertEqual(response.status_code, status.HTTP_422_UNPROCESSABLE_ENTITY)
self.assertEqual(response.data, 'invalid enrollment record')
def test_program_unauthorized(self):
student = UserFactory.create(password='password')
self.client.login(username=student.username, password='password')
request_data = [self.student_enrollment('enrolled')]
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
response = self.request(self.get_url(), json.dumps(request_data), content_type='application/json')
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
def test_program_not_found(self):
post_data = [self.student_enrollment('enrolled')]
nonexistant_uuid = uuid4()
response = self.request(
self.get_url(program_uuid=nonexistant_uuid),
json.dumps(post_data),
content_type='application/json'
)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
@ddt.data(
[{'status': 'pending'}],
[{'status': 'not-a-status'}],
[{'status': 'pending'}, {'status': 'pending'}],
)
def test_no_student_key(self, bad_records):
program_uuid = uuid4()
url = reverse('programs_api:v1:program_enrollments', args=[program_uuid])
enrollments = [self.student_enrollment('enrolled', '001', True)]
enrollments.extend(bad_records)
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
response = self.request(url, json.dumps(enrollments), content_type='application/json')
self.assertEqual(422, response.status_code)
self.assertEqual('invalid enrollment record', response.data)
def test_extra_field(self):
self.student_enrollment('pending', 'learner-01', prepare_student=True)
enrollment = self.student_enrollment('enrolled', 'learner-01')
enrollment['favorite_pokemon'] = 'bulbasaur'
enrollments = [enrollment]
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
with mock.patch(
'lms.djangoapps.program_enrollments.api.v1.views.get_user_by_program_id',
autospec=True,
return_value=None
):
url = self.get_url(program_uuid=self.program_uuid)
response = self.request(url, json.dumps(enrollments), content_type='application/json')
self.assertEqual(self.success_status, response.status_code)
self.assertDictEqual(
response.data,
{'learner-01': 'enrolled'}
)
@ddt.ddt
class ProgramEnrollmentViewPostTests(BaseProgramEnrollmentWriteTestsMixin, APITestCase):
"""
Tests for the ProgramEnrollment view POST method.
"""
add_uuid = True
success_status = status.HTTP_201_CREATED
success_status = 201
def setUp(self):
super(ProgramEnrollmentViewPostTests, self).setUp()
self.request = self.client.post
global_staff = GlobalStaffFactory.create(username='global-staff', password='password')
self.client.login(username=global_staff.username, password='password')
def tearDown(self):
super(ProgramEnrollmentViewPostTests, self).tearDown()
ProgramEnrollment.objects.all().delete()
def test_successful_program_enrollments_no_existing_user(self):
program_key = uuid4()
statuses = ['pending', 'enrolled', 'pending']
external_user_keys = ['abc1', 'efg2', 'hij3']
curriculum_uuid = uuid4()
curriculum_uuids = [curriculum_uuid, curriculum_uuid, uuid4()]
post_data = [
{
REQUEST_STUDENT_KEY: e,
'status': s,
'curriculum_uuid': str(c)
}
for e, s, c in zip(external_user_keys, statuses, curriculum_uuids)
]
url = reverse('programs_api:v1:program_enrollments', args=[program_key])
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
with mock.patch(
'lms.djangoapps.program_enrollments.api.v1.views.get_user_by_program_id',
autospec=True,
return_value=None
):
response = self.client.post(url, json.dumps(post_data), content_type='application/json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
for i in range(3):
enrollment = ProgramEnrollment.objects.get(external_user_key=external_user_keys[i])
self.assertEqual(enrollment.external_user_key, external_user_keys[i])
self.assertEqual(enrollment.program_uuid, program_key)
self.assertEqual(enrollment.status, statuses[i])
self.assertEqual(enrollment.curriculum_uuid, curriculum_uuids[i])
self.assertEqual(enrollment.user, None)
def test_successful_program_enrollments_existing_user(self):
program_key = uuid4()
curriculum_uuid = uuid4()
post_data = [
{
'status': 'enrolled',
REQUEST_STUDENT_KEY: 'abc1',
'curriculum_uuid': str(curriculum_uuid)
}
]
user = User.objects.create_user('test_user', 'test@example.com', 'password')
url = reverse('programs_api:v1:program_enrollments', args=[program_key])
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
with mock.patch(
'lms.djangoapps.program_enrollments.api.v1.views.get_user_by_program_id',
autospec=True,
return_value=user
):
response = self.client.post(url, json.dumps(post_data), content_type='application/json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
enrollment = ProgramEnrollment.objects.get(external_user_key='abc1')
self.assertEqual(enrollment.external_user_key, 'abc1')
self.assertEqual(enrollment.program_uuid, program_key)
self.assertEqual(enrollment.status, 'enrolled')
self.assertEqual(enrollment.curriculum_uuid, curriculum_uuid)
self.assertEqual(enrollment.user, user)
def test_program_enrollments_no_idp(self):
program_key = uuid4()
curriculum_uuid = uuid4()
post_data = [
{
'status': 'enrolled',
REQUEST_STUDENT_KEY: 'abc{}'.format(i),
'curriculum_uuid': str(curriculum_uuid)
} for i in range(3)
]
url = reverse('programs_api:v1:program_enrollments', args=[program_key])
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
with mock.patch(
'lms.djangoapps.program_enrollments.api.v1.views.get_user_by_program_id',
autospec=True,
side_effect=ProviderDoesNotExistException()
):
response = self.client.post(url, json.dumps(post_data), content_type='application/json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
for i in range(3):
enrollment = ProgramEnrollment.objects.get(external_user_key='abc{}'.format(i))
self.assertEqual(enrollment.program_uuid, program_key)
self.assertEqual(enrollment.status, 'enrolled')
self.assertEqual(enrollment.curriculum_uuid, curriculum_uuid)
self.assertIsNone(enrollment.user)
@ddt.ddt
class ProgramEnrollmentViewPatchTests(BaseProgramEnrollmentWriteTestsMixin, APITestCase):
"""
Tests for the ProgramEnrollment view PATCH method.
"""
add_uuid = False
success_status = status.HTTP_200_OK
def setUp(self):
super(ProgramEnrollmentViewPatchTests, self).setUp()
self.request = self.client.patch
self.curriculum_uuid = 'aaaaaaaa-1111-2222-3333-444444444444'
self.other_curriculum_uuid = 'bbbbbbbb-1111-2222-3333-444444444444'
self.course_id = CourseKey.from_string('course-v1:edX+ToyX+Toy_Course')
_ = CourseOverviewFactory.create(id=self.course_id)
self.password = 'password'
self.student = UserFactory.create(username='student', password=self.password)
self.global_staff = GlobalStaffFactory.create(username='global-staff', password=self.password)
self.client.login(username=self.global_staff.username, password=self.password)
def prepare_student(self, enrollment):
ProgramEnrollment.objects.create(
program_uuid=self.program_uuid,
curriculum_uuid=self.curriculum_uuid,
user=None,
status='pending',
external_user_key=enrollment[REQUEST_STUDENT_KEY],
)
def test_successfully_patched_program_enrollment(self):
enrollments = {}
for i in range(4):
user_key = 'user-{}'.format(i)
instance = ProgramEnrollment.objects.create(
program_uuid=self.program_uuid,
curriculum_uuid=self.curriculum_uuid,
user=None,
status='pending',
external_user_key=user_key,
)
enrollments[user_key] = instance
post_data = [
{REQUEST_STUDENT_KEY: 'user-1', 'status': 'canceled'},
{REQUEST_STUDENT_KEY: 'user-2', 'status': 'suspended'},
{REQUEST_STUDENT_KEY: 'user-3', 'status': 'enrolled'},
]
url = reverse('programs_api:v1:program_enrollments', args=[self.program_uuid])
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
response = self.client.patch(url, json.dumps(post_data), content_type='application/json')
for enrollment in enrollments.values():
enrollment.refresh_from_db()
expected_statuses = {
'user-0': 'pending',
'user-1': 'canceled',
'user-2': 'suspended',
'user-3': 'enrolled',
}
for user_key, enrollment in enrollments.items():
assert expected_statuses[user_key] == enrollment.status
expected_response = {
'user-1': 'canceled',
'user-2': 'suspended',
'user-3': 'enrolled',
}
assert status.HTTP_200_OK == response.status_code
assert expected_response == response.data
def test_duplicate_enrollment_record_changed(self):
enrollments = {}
for i in range(4):
user_key = 'user-{}'.format(i)
instance = ProgramEnrollment.objects.create(
program_uuid=self.program_uuid,
curriculum_uuid=self.curriculum_uuid,
user=None,
status='pending',
external_user_key=user_key,
)
enrollments[user_key] = instance
patch_data = [
self.student_enrollment('enrolled', 'user-1'),
self.student_enrollment('enrolled', 'user-2'),
self.student_enrollment('enrolled', 'user-1'),
]
url = reverse('programs_api:v1:program_enrollments', args=[self.program_uuid])
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
response = self.client.patch(url, json.dumps(patch_data), content_type='application/json')
for enrollment in enrollments.values():
enrollment.refresh_from_db()
expected_statuses = {
'user-0': 'pending',
'user-1': 'pending',
'user-2': 'enrolled',
'user-3': 'pending',
}
for user_key, enrollment in enrollments.items():
assert expected_statuses[user_key] == enrollment.status
self.assertEqual(response.status_code, status.HTTP_207_MULTI_STATUS)
self.assertEqual(response.data, {
'user-1': 'duplicated',
'user-2': 'enrolled',
})
def test_partially_valid_enrollment_record_changed(self):
enrollments = {}
for i in range(4):
user_key = 'user-{}'.format(i)
instance = ProgramEnrollment.objects.create(
program_uuid=self.program_uuid,
curriculum_uuid=self.curriculum_uuid,
user=None,
status='pending',
external_user_key=user_key,
)
enrollments[user_key] = instance
patch_data = [
self.student_enrollment('new', 'user-1'),
self.student_enrollment('canceled', 'user-3'),
self.student_enrollment('enrolled', 'user-who-is-not-in-program'),
]
url = reverse('programs_api:v1:program_enrollments', args=[self.program_uuid])
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
response = self.client.patch(url, json.dumps(patch_data), content_type='application/json')
for enrollment in enrollments.values():
enrollment.refresh_from_db()
expected_statuses = {
'user-0': 'pending',
'user-1': 'pending',
'user-2': 'pending',
'user-3': 'canceled',
}
for user_key, enrollment in enrollments.items():
assert expected_statuses[user_key] == enrollment.status
self.assertEqual(response.status_code, status.HTTP_207_MULTI_STATUS)
self.assertEqual(response.data, {
'user-1': 'invalid-status',
'user-3': 'canceled',
'user-who-is-not-in-program': 'not-in-program',
})
@ddt.ddt
class ProgramEnrollmentViewPutTests(BaseProgramEnrollmentWriteTestsMixin, APITestCase):
"""
Tests for the ProgramEnrollment view PATCH method.
"""
add_uuid = True
success_status = status.HTTP_200_OK
def setUp(self):
super(ProgramEnrollmentViewPutTests, self).setUp()
self.request = self.client.put
self.program_uuid = '00000000-1111-2222-3333-444444444444'
self.curriculum_uuid = 'aaaaaaaa-1111-2222-3333-444444444444'
self.global_staff = GlobalStaffFactory.create(username='global-staff', password='password')
self.client.login(username=self.global_staff.username, password='password')
patch_get_user = mock.patch(
'lms.djangoapps.program_enrollments.api.v1.views.get_user_by_program_id',
autospec=True,
return_value=None
)
self.mock_get_user = patch_get_user.start()
self.addCleanup(patch_get_user.stop)
def prepare_student(self, enrollment):
ProgramEnrollment.objects.create(
program_uuid=self.program_uuid,
curriculum_uuid=self.curriculum_uuid,
user=None,
status='pending',
external_user_key=enrollment[REQUEST_STUDENT_KEY],
)
@ddt.data(True, False)
def test_all_create_or_modify(self, create_users):
request_data = [
self.student_enrollment(ProgramStatuses.ENROLLED)
for _ in range(5)
]
if create_users:
for enrollment in request_data:
ProgramEnrollmentFactory(
program_uuid=self.program_uuid,
status=ProgramStatuses.PENDING,
external_user_key=enrollment[REQUEST_STUDENT_KEY],
)
url = self.get_url(program_uuid=self.program_uuid)
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
response = self.client.put(url, json.dumps(request_data), content_type='application/json')
self.assertEqual(self.success_status, response.status_code)
self.assertEqual(5, len(response.data))
for response_status in response.data.values():
self.assertEqual(response_status, ProgramStatuses.ENROLLED)
def test_half_create_modify(self):
request_data = [
self.student_enrollment(ProgramStatuses.ENROLLED, 'learner-01'),
self.student_enrollment(ProgramStatuses.ENROLLED, 'learner-02'),
self.student_enrollment(ProgramStatuses.ENROLLED, 'learner-03'),
self.student_enrollment(ProgramStatuses.ENROLLED, 'learner-04'),
]
ProgramEnrollmentFactory(
program_uuid=self.program_uuid,
status=ProgramStatuses.PENDING,
external_user_key='learner-03',
)
ProgramEnrollmentFactory(
program_uuid=self.program_uuid,
status=ProgramStatuses.PENDING,
external_user_key='learner-04',
)
url = self.get_url(program_uuid=self.program_uuid)
with mock.patch('lms.djangoapps.program_enrollments.api.v1.views.get_programs', autospec=True):
response = self.client.put(url, json.dumps(request_data), content_type='application/json')
self.assertEqual(self.success_status, response.status_code)
self.assertEqual(4, len(response.data))
for response_status in response.data.values():
self.assertEqual(response_status, ProgramStatuses.ENROLLED)
@ddt.ddt
class ProgramCourseEnrollmentOverviewViewTests(ProgramCacheTestCaseMixin, SharedModuleStoreTestCase, APITestCase):
"""
Tests for the ProgramCourseEnrollmentOverview view GET method.
"""
@classmethod
def setUpClass(cls):
super(ProgramCourseEnrollmentOverviewViewTests, cls).setUpClass()
cls.program_uuid = '00000000-1111-2222-3333-444444444444'
cls.curriculum_uuid = 'aaaaaaaa-1111-2222-3333-444444444444'
cls.other_curriculum_uuid = 'bbbbbbbb-1111-2222-3333-444444444444'
cls.course_id = CourseKey.from_string('course-v1:edX+ToyX+Toy_Course')
cls.course_run = CourseRunFactory.create(key=text_type(cls.course_id))
cls.course = CourseFactory.create(course_runs=[cls.course_run])
cls.password = 'password'
cls.student = UserFactory.create(username='student', password=cls.password)
# only freeze time when defining these values and not on the whole test case
# as test_multiple_enrollments_all_enrolled relies on actual differences in modified datetimes
with freeze_time('2019-01-01'):
cls.yesterday = datetime.utcnow() - timedelta(1)
cls.tomorrow = datetime.utcnow() + timedelta(1)
cls.relative_certificate_download_url = '/download-the-certificates'
cls.absolute_certificate_download_url = 'http://www.certificates.com/'
def setUp(self):
super(ProgramCourseEnrollmentOverviewViewTests, self).setUp()
# create program enrollment
self.program_enrollment = ProgramEnrollmentFactory.create(
program_uuid=self.program_uuid,
curriculum_uuid=self.curriculum_uuid,
user=self.student,
)
# create course enrollment
self.course_enrollment = CourseEnrollmentFactory.create(
course_id=self.course_id,
user=self.student,
mode=CourseMode.MASTERS,
)
# create course overview
self.course_overview = CourseOverviewFactory.create(
id=self.course_id,
start=self.yesterday,
end=self.tomorrow,
)
# create program course enrollment
self.program_course_enrollment = ProgramCourseEnrollmentFactory.create(
program_enrollment=self.program_enrollment,
course_enrollment=self.course_enrollment,
course_key=self.course_id,
status='active',
)
# create program
self.program = self.setup_catalog_cache(self.program_uuid, 'organization_key')
self.program['curricula'][0]['courses'].append(self.course)
self.set_program_in_catalog_cache(self.program_uuid, self.program)
def create_generated_certificate(self, download_url=None):
return GeneratedCertificateFactory.create(
user=self.student,
course_id=self.course_id,
status=CertificateStatuses.downloadable,
mode='verified',
download_url=(download_url or self.relative_certificate_download_url),
grade="0.88",
verify_uuid=uuid4(),
)
def get_url(self, program_uuid=None):
""" Returns the primary URL requested by the test case. """
kwargs = {'program_uuid': program_uuid or self.program_uuid}
return reverse('programs_api:v1:program_course_enrollments_overview', kwargs=kwargs)
def test_401_if_anonymous(self):
response = self.client.get(self.get_url(self.program_uuid))
assert status.HTTP_401_UNAUTHORIZED == response.status_code
def test_404_if_no_program_with_key(self):
self.client.login(username=self.student.username, password=self.password)
self.set_program_in_catalog_cache(self.program_uuid, None)
response = self.client.get(self.get_url(self.program_uuid))
assert status.HTTP_404_NOT_FOUND == response.status_code
def test_403_if_not_enrolled_in_program(self):
# delete program enrollment
ProgramEnrollment.objects.all().delete()
self.client.login(username=self.student.username, password=self.password)
response = self.client.get(self.get_url(self.program_uuid))
assert status.HTTP_403_FORBIDDEN == response.status_code
def _add_new_course_to_program(self, course_run_key, program):
"""
Helper method to create another course, an overview for it,
add it to the program, and re-load the cache.
"""
other_course_run = CourseRunFactory.create(key=text_type(course_run_key))
other_course = CourseFactory.create(course_runs=[other_course_run])
program['courses'].append(other_course)
self.set_program_in_catalog_cache(program['uuid'], program)
CourseOverviewFactory.create(
id=course_run_key,
start=self.yesterday,
)
@ddt.data(False, True)
def test_multiple_enrollments_all_enrolled(self, other_enrollment_active):
other_course_key = CourseKey.from_string('course-v1:edX+ToyX+Other_Course')
self._add_new_course_to_program(other_course_key, self.program)
# add a second course enrollment, which doesn't need a ProgramCourseEnrollment
# to be returned.
other_enrollment = CourseEnrollmentFactory.create(
course_id=other_course_key,
user=self.student,
mode=CourseMode.VERIFIED,
)
if not other_enrollment_active:
other_enrollment.deactivate()
self.client.login(username=self.student.username, password=self.password)
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
actual_course_run_ids = {run['course_run_id'] for run in response.data['course_runs']}
expected_course_run_ids = {text_type(self.course_id)}
if other_enrollment_active:
expected_course_run_ids.add(text_type(other_course_key))
self.assertEqual(expected_course_run_ids, actual_course_run_ids)
_GET_RESUME_URL = 'lms.djangoapps.program_enrollments.api.v1.views.get_resume_urls_for_enrollments'
@mock.patch(_GET_RESUME_URL)
def test_blank_resume_url_omitted(self, mock_get_resume_urls):
self.client.login(username=self.student.username, password=self.password)
mock_get_resume_urls.return_value = {self.course_id: ''}
response = self.client.get(self.get_url(self.program_uuid))
self.assertNotIn('resume_course_run_url', response.data['course_runs'][0])
@mock.patch(_GET_RESUME_URL)
def test_relative_resume_url_becomes_absolute(self, mock_get_resume_urls):
self.client.login(username=self.student.username, password=self.password)
resume_url = '/resume-here'
mock_get_resume_urls.return_value = {self.course_id: resume_url}
response = self.client.get(self.get_url(self.program_uuid))
response_resume_url = response.data['course_runs'][0]['resume_course_run_url']
self.assertTrue(response_resume_url.startswith("http://testserver"))
self.assertTrue(response_resume_url.endswith(resume_url))
@mock.patch(_GET_RESUME_URL)
def test_absolute_resume_url_stays_absolute(self, mock_get_resume_urls):
self.client.login(username=self.student.username, password=self.password)
resume_url = 'http://www.resume.com/'
mock_get_resume_urls.return_value = {self.course_id: resume_url}
response = self.client.get(self.get_url(self.program_uuid))
response_resume_url = response.data['course_runs'][0]['resume_course_run_url']
self.assertEqual(response_resume_url, resume_url)
def test_no_url_without_certificate(self):
self.client.login(username=self.student.username, password=self.password)
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertNotIn('certificate_download_url', response.data['course_runs'][0])
def test_relative_certificate_url_becomes_absolute(self):
self.client.login(username=self.student.username, password=self.password)
self.create_generated_certificate(
download_url=self.relative_certificate_download_url
)
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
response_url = response.data['course_runs'][0]['certificate_download_url']
self.assertTrue(response_url.startswith("http://testserver"))
self.assertTrue(response_url.endswith(self.relative_certificate_download_url))
def test_absolute_certificate_url_stays_absolute(self):
self.client.login(username=self.student.username, password=self.password)
self.create_generated_certificate(
download_url=self.absolute_certificate_download_url
)
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
response_url = response.data['course_runs'][0]['certificate_download_url']
self.assertEqual(response_url, self.absolute_certificate_download_url)
def test_no_due_dates(self):
self.client.login(username=self.student.username, password=self.password)
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
assert [] == response.data['course_runs'][0]['due_dates']
def test_due_dates(self):
course = ModulestoreCourseFactory.create(
org="edX",
course="ToyX",
run="Toy_Course",
)
section_1 = ItemFactory.create(
category='chapter',
start=self.yesterday,
due=self.tomorrow,
parent=course,
display_name='section 1'
)
subsection_1 = ItemFactory.create(
category='sequential',
due=self.tomorrow,
parent=section_1,
display_name='subsection 1'
)
subsection_2 = ItemFactory.create(
category='sequential',
due=self.tomorrow - timedelta(1),
parent=section_1,
display_name='subsection 2'
)
subsection_3 = ItemFactory.create(
category='sequential',
parent=section_1,
display_name='subsection 3'
)
unit_1 = ItemFactory.create(
category='vertical',
due=self.tomorrow + timedelta(2),
parent=subsection_3,
display_name='unit_1'
)
with mock.patch('lms.djangoapps.program_enrollments.api.api.get_dates_for_course') as mock_get_dates:
mock_get_dates.return_value = {
(section_1.location, 'due'): section_1.due,
(section_1.location, 'start'): section_1.start,
(subsection_1.location, 'due'): subsection_1.due,
(subsection_2.location, 'due'): subsection_2.due,
(unit_1.location, 'due'): unit_1.due,
}
self.client.login(username=self.student.username, password=self.password)
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
block_data = [
{
'name': section_1.display_name,
'url': ('http://testserver/courses/course-v1:edX+ToyX+Toy_Course/'
'jump_to/i4x://edX/ToyX/chapter/section_1'),
'date': '2019-01-02T00:00:00Z',
},
{
'name': subsection_1.display_name,
'url': ('http://testserver/courses/course-v1:edX+ToyX+Toy_Course/'
'jump_to/i4x://edX/ToyX/sequential/subsection_1'),
'date': '2019-01-02T00:00:00Z',
},
{
'name': subsection_2.display_name,
'url': ('http://testserver/courses/course-v1:edX+ToyX+Toy_Course/'
'jump_to/i4x://edX/ToyX/sequential/subsection_2'),
'date': '2019-01-01T00:00:00Z',
},
{
'name': unit_1.display_name,
'url': ('http://testserver/courses/course-v1:edX+ToyX+Toy_Course/'
'jump_to/i4x://edX/ToyX/vertical/unit_1'),
'date': '2019-01-04T00:00:00Z',
},
]
due_dates = response.data['course_runs'][0]['due_dates']
for block in block_data:
self.assertIn(block, due_dates)
@mock.patch.object(CourseOverview, 'has_ended')
def test_course_run_status_instructor_paced_completed(self, mock_has_ended):
self.client.login(username=self.student.username, password=self.password)
# set as instructor paced
self.course_overview.self_paced = False
self.course_overview.save()
mock_has_ended.return_value = True
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertEqual(CourseRunProgressStatuses.COMPLETED, response.data['course_runs'][0]['course_run_status'])
@mock.patch.object(CourseOverview, 'has_ended')
@mock.patch.object(CourseOverview, 'has_started')
def test_course_run_status_instructor_paced_in_progress(self, mock_has_started, mock_has_ended):
self.client.login(username=self.student.username, password=self.password)
# set as instructor paced
self.course_overview.self_paced = False
self.course_overview.save()
mock_has_started.return_value = True
mock_has_ended.return_value = False
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertEqual(CourseRunProgressStatuses.IN_PROGRESS, response.data['course_runs'][0]['course_run_status'])
@mock.patch.object(CourseOverview, 'has_ended')
@mock.patch.object(CourseOverview, 'has_started')
def test_course_run_status_instructor_paced_upcoming(self, mock_has_started, mock_has_ended):
self.client.login(username=self.student.username, password=self.password)
# set as instructor paced
self.course_overview.self_paced = False
self.course_overview.save()
mock_has_started.return_value = False
mock_has_ended.return_value = False
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertEqual(CourseRunProgressStatuses.UPCOMING, response.data['course_runs'][0]['course_run_status'])
@mock.patch.object(CourseOverview, 'has_ended')
def test_course_run_status_self_paced_completed(self, mock_has_ended):
self.client.login(username=self.student.username, password=self.password)
# set as self paced
self.course_overview.self_paced = True
self.course_overview.save()
# course run has ended
mock_has_ended.return_value = True
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertEqual(CourseRunProgressStatuses.COMPLETED, response.data['course_runs'][0]['course_run_status'])
# course run has not ended and user has earned a passing certificate more than 30 days ago
certificate = self.create_generated_certificate()
certificate.created_date = datetime.utcnow() - timedelta(30)
certificate.save()
mock_has_ended.return_value = False
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertEqual(CourseRunProgressStatuses.COMPLETED, response.data['course_runs'][0]['course_run_status'])
# course run has ended and user has earned a passing certificate
mock_has_ended.return_value = True
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertEqual(CourseRunProgressStatuses.COMPLETED, response.data['course_runs'][0]['course_run_status'])
@mock.patch.object(CourseOverview, 'has_ended')
@mock.patch.object(CourseOverview, 'has_started')
def test_course_run_status_self_paced_in_progress(self, mock_has_started, mock_has_ended):
self.client.login(username=self.student.username, password=self.password)
# set as self paced
self.course_overview.self_paced = True
self.course_overview.save()
# course run has started and has not ended
mock_has_started.return_value = True
mock_has_ended.return_value = False
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertEqual(CourseRunProgressStatuses.IN_PROGRESS, response.data['course_runs'][0]['course_run_status'])
# course run has not ended and user has earned a passing certificate fewer than 30 days ago
certificate = self.create_generated_certificate()
certificate.created_date = datetime.utcnow() - timedelta(5)
certificate.save()
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertEqual(CourseRunProgressStatuses.IN_PROGRESS, response.data['course_runs'][0]['course_run_status'])
@mock.patch.object(CourseOverview, 'has_ended')
@mock.patch.object(CourseOverview, 'has_started')
def test_course_run_status_self_paced_upcoming(self, mock_has_started, mock_has_ended):
self.client.login(username=self.student.username, password=self.password)
# set as self paced
self.course_overview.self_paced = True
self.course_overview.save()
# course run has not started and has not ended
mock_has_started.return_value = False
mock_has_ended.return_value = False
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertEqual(CourseRunProgressStatuses.UPCOMING, response.data['course_runs'][0]['course_run_status'])
def test_course_run_url(self):
self.client.login(username=self.student.username, password=self.password)
course_run_url = 'http://testserver/courses/{}/course/'.format(text_type(self.course_id))
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertEqual(course_run_url, response.data['course_runs'][0]['course_run_url'])
def test_course_run_dates(self):
self.client.login(username=self.student.username, password=self.password)
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
course_run_overview = response.data['course_runs'][0]
self.assertEqual(course_run_overview['start_date'], '2018-12-31T00:00:00Z')
self.assertEqual(course_run_overview['end_date'], '2019-01-02T00:00:00Z')
# course run end date may not exist
self.course_overview.end = None
self.course_overview.save()
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertEqual(response.data['course_runs'][0]['end_date'], None)
def test_course_run_id_and_display_name(self):
self.client.login(username=self.student.username, password=self.password)
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
course_run_overview = response.data['course_runs'][0]
self.assertEqual(course_run_overview['course_run_id'], text_type(self.course_id))
self.assertEqual(course_run_overview['display_name'], "{} Course".format(text_type(self.course_id)))
def test_emails_enabled(self):
self.client.login(username=self.student.username, password=self.password)
# by default, BulkEmailFlag is not enabled, so 'emails_enabled' won't be in the response
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertNotIn('emails_enabled', response.data['course_runs'][0])
with mock.patch.object(BulkEmailFlag, 'feature_enabled', return_value=True):
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertTrue(response.data['course_runs'][0]['emails_enabled'])
Optout.objects.create(
user=self.student,
course_id=self.course_id
)
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertFalse(response.data['course_runs'][0]['emails_enabled'])
def test_micromasters_title(self):
self.client.login(username=self.student.username, password=self.password)
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertNotIn('micromasters_title', response.data['course_runs'][0])
self.program['type'] = 'MicroMasters'
# update the program in the catalog cache
self.set_program_in_catalog_cache(self.program_uuid, self.program)
response = self.client.get(self.get_url(self.program_uuid))
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertIn('micromasters_title', response.data['course_runs'][0])
class ProgramCourseGradeListTest(ProgramEnrollmentDataMixin, ListViewTestMixin, APITestCase):
"""
Tests for GET calls to the Program Course Grades API.
"""
view_name = 'programs_api:v1:program_course_grades'
@staticmethod
def mock_course_grade(percent=75.0, passed=True, letter_grade='B'):
return mock.MagicMock(percent=percent, passed=passed, letter_grade=letter_grade)
@mock.patch('lms.djangoapps.program_enrollments.api.v1.views.CourseGradeFactory')
def test_204_no_grades_to_return(self, mock_course_grade_factory):
mock_course_grade_factory.return_value.iter.return_value = []
self.log_in_staff()
url = self.get_url(program_uuid=self.program_uuid, course_id=self.course_key)
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.assertEqual(response.data['results'], [])
def test_401_if_unauthenticated(self):
url = self.get_url(program_uuid=self.program_uuid, course_id=self.course_key)
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
def test_403_if_not_staff(self):
self.log_in_non_staff()
url = self.get_url(program_uuid=self.program_uuid, course_id=self.course_key)
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
def test_404_not_found(self):
fake_program_uuid = self.program_uuid_tmpl.format(99)
self.log_in_staff()
url = self.get_url(program_uuid=fake_program_uuid, course_id=self.course_key)
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
@mock.patch('lms.djangoapps.program_enrollments.api.v1.views.CourseGradeFactory')
def test_200_grades_with_no_exceptions(self, mock_course_grade_factory):
other_student = UserFactory.create(username='other_student')
self.create_program_and_course_enrollments('student-key', user=self.student)
self.create_program_and_course_enrollments('other-student-key', user=other_student)
mock_course_grades = [
(self.student, self.mock_course_grade(), None),
(other_student, self.mock_course_grade(percent=40.0, passed=False, letter_grade='F'), None),
]
mock_course_grade_factory.return_value.iter.return_value = mock_course_grades
self.log_in_staff()
url = self.get_url(program_uuid=self.program_uuid, course_id=self.course_key)
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
expected_results = [
{
'student_key': 'student-key',
'passed': True,
'percent': 75.0,
'letter_grade': 'B',
},
{
'student_key': 'other-student-key',
'passed': False,
'percent': 40.0,
'letter_grade': 'F',
},
]
self.assertEqual(response.data['results'], expected_results)
@mock.patch('lms.djangoapps.program_enrollments.api.v1.views.CourseGradeFactory')
def test_207_grades_with_some_exceptions(self, mock_course_grade_factory):
other_student = UserFactory.create(username='other_student')
self.create_program_and_course_enrollments('student-key', user=self.student)
self.create_program_and_course_enrollments('other-student-key', user=other_student)
mock_course_grades = [
(self.student, None, Exception('Bad Data')),
(other_student, self.mock_course_grade(percent=40.0, passed=False, letter_grade='F'), None),
]
mock_course_grade_factory.return_value.iter.return_value = mock_course_grades
self.log_in_staff()
url = self.get_url(program_uuid=self.program_uuid, course_id=self.course_key)
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_207_MULTI_STATUS)
expected_results = [
{
'student_key': 'student-key',
'error': 'Bad Data',
},
{
'student_key': 'other-student-key',
'passed': False,
'percent': 40.0,
'letter_grade': 'F',
},
]
self.assertEqual(response.data['results'], expected_results)
@mock.patch('lms.djangoapps.program_enrollments.api.v1.views.CourseGradeFactory')
def test_422_grades_with_only_exceptions(self, mock_course_grade_factory):
other_student = UserFactory.create(username='other_student')
self.create_program_and_course_enrollments('student-key', user=self.student)
self.create_program_and_course_enrollments('other-student-key', user=other_student)
mock_course_grades = [
(self.student, None, Exception('Bad Data')),
(other_student, None, Exception('Timeout')),
]
mock_course_grade_factory.return_value.iter.return_value = mock_course_grades
self.log_in_staff()
url = self.get_url(program_uuid=self.program_uuid, course_id=self.course_key)
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_422_UNPROCESSABLE_ENTITY)
expected_results = [
{
'student_key': 'student-key',
'error': 'Bad Data',
},
{
'student_key': 'other-student-key',
'error': 'Timeout',
},
]
self.assertEqual(response.data['results'], expected_results)
class EnrollmentDataResetViewTests(ProgramCacheTestCaseMixin, APITestCase):
""" Tests endpoint for resetting enrollments in integration environments """
FEATURES_WITH_ENABLED = settings.FEATURES.copy()
FEATURES_WITH_ENABLED[ENABLE_ENROLLMENT_RESET_FLAG] = True
reset_enrollments_cmd = 'reset_enrollment_data'
reset_users_cmd = 'remove_social_auth_users'
def setUp(self):
super(EnrollmentDataResetViewTests, self).setUp()
self.start_cache_isolation()
self.organization = OrganizationFactory(short_name='uox')
self.provider = SAMLProviderConfigFactory(organization=self.organization)
self.global_staff = GlobalStaffFactory.create(username='global-staff', password='password')
self.client.login(username=self.global_staff.username, password='password')
def request(self, organization):
return self.client.post(
reverse('programs_api:v1:reset_enrollment_data'),
{'organization': organization},
format='json',
)
def tearDown(self):
self.end_cache_isolation()
super(EnrollmentDataResetViewTests, self).tearDown()
@mock.patch('lms.djangoapps.program_enrollments.api.v1.views.call_command', autospec=True)
def test_feature_disabled_by_default(self, mock_call_command):
response = self.request(self.organization.short_name)
self.assertEqual(response.status_code, status.HTTP_501_NOT_IMPLEMENTED)
mock_call_command.assert_has_calls([])
@override_settings(FEATURES=FEATURES_WITH_ENABLED)
@mock.patch('lms.djangoapps.program_enrollments.api.v1.views.call_command', autospec=True)
def test_403_for_non_staff(self, mock_call_command):
student = UserFactory.create(username='student', password='password')
self.client.login(username=student.username, password='password')
response = self.request(self.organization.short_name)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
mock_call_command.assert_has_calls([])
@override_settings(FEATURES=FEATURES_WITH_ENABLED)
@mock.patch('lms.djangoapps.program_enrollments.api.v1.views.call_command', autospec=True)
def test_reset(self, mock_call_command):
programs = [str(uuid4()), str(uuid4())]
self.set_org_in_catalog_cache(self.organization, programs)
response = self.request(self.organization.short_name)
self.assertEqual(response.status_code, status.HTTP_200_OK)
mock_call_command.assert_has_calls([
mock.call(self.reset_users_cmd, self.provider.slug, force=True),
mock.call(self.reset_enrollments_cmd, ','.join(programs), force=True),
])
@override_settings(FEATURES=FEATURES_WITH_ENABLED)
@mock.patch('lms.djangoapps.program_enrollments.api.v1.views.call_command', autospec=True)
def test_reset_without_idp(self, mock_call_command):
organization = OrganizationFactory()
programs = [str(uuid4()), str(uuid4())]
self.set_org_in_catalog_cache(organization, programs)
response = self.request(organization.short_name)
self.assertEqual(response.status_code, status.HTTP_200_OK)
mock_call_command.assert_has_calls([
mock.call(self.reset_enrollments_cmd, ','.join(programs), force=True),
])
@override_settings(FEATURES=FEATURES_WITH_ENABLED)
@mock.patch('lms.djangoapps.program_enrollments.api.v1.views.call_command', autospec=True)
def test_organization_not_found(self, mock_call_command):
response = self.request('yyz')
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
mock_call_command.assert_has_calls([])
@override_settings(FEATURES=FEATURES_WITH_ENABLED)
@mock.patch('lms.djangoapps.program_enrollments.api.v1.views.call_command', autospec=True)
def test_no_programs_doesnt_break(self, mock_call_command):
programs = []
self.set_org_in_catalog_cache(self.organization, programs)
response = self.request(self.organization.short_name)
self.assertEqual(response.status_code, status.HTTP_200_OK)
mock_call_command.assert_has_calls([
mock.call(self.reset_users_cmd, self.provider.slug, force=True),
])
@override_settings(FEATURES=FEATURES_WITH_ENABLED)
@mock.patch('lms.djangoapps.program_enrollments.api.v1.views.call_command', autospec=True)
def test_missing_body_content(self, mock_call_command):
response = self.client.post(
reverse('programs_api:v1:reset_enrollment_data'),
format='json',
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
mock_call_command.assert_has_calls([])

63
lms/djangoapps/program_enrollments/api/v1/urls.py
View File

@@ -1,63 +0,0 @@
""" Program Enrollments API v1 URLs. """
from __future__ import absolute_import
from django.conf.urls import url
from lms.djangoapps.program_enrollments.api.v1.constants import PROGRAM_UUID_PATTERN
from lms.djangoapps.program_enrollments.api.v1.views import (
EnrollmentDataResetView,
ProgramEnrollmentsView,
ProgramCourseEnrollmentsView,
ProgramCourseGradesView,
ProgramCourseEnrollmentOverviewView,
UserProgramReadOnlyAccessView,
)
from openedx.core.constants import COURSE_ID_PATTERN
app_name = 'lms.djangoapps.program_enrollments'
urlpatterns = [
url(
r'^programs/enrollments/$',
UserProgramReadOnlyAccessView.as_view(),
name='learner_program_enrollments'
),
url(
r'^programs/readonly_access/$',
UserProgramReadOnlyAccessView.as_view(),
name='user_program_readonly_access'
),
url(
r'^programs/{program_uuid}/enrollments/$'.format(program_uuid=PROGRAM_UUID_PATTERN),
ProgramEnrollmentsView.as_view(),
name='program_enrollments'
),
url(
r'^programs/{program_uuid}/courses/{course_id}/enrollments/'.format(
program_uuid=PROGRAM_UUID_PATTERN,
course_id=COURSE_ID_PATTERN
),
ProgramCourseEnrollmentsView.as_view(),
name="program_course_enrollments"
),
url(
r'^programs/{program_uuid}/courses/{course_id}/grades/'.format(
program_uuid=PROGRAM_UUID_PATTERN,
course_id=COURSE_ID_PATTERN
),
ProgramCourseGradesView.as_view(),
name="program_course_grades"
),
url(
r'^programs/{program_uuid}/overview/'.format(
program_uuid=PROGRAM_UUID_PATTERN,
),
ProgramCourseEnrollmentOverviewView.as_view(),
name="program_course_enrollments_overview"
),
url(
r'^integration-reset',
EnrollmentDataResetView.as_view(),
name="reset_enrollment_data",
)
]

1346
lms/djangoapps/program_enrollments/api/v1/views.py
View File

@@ -1,1346 +0,0 @@
# -*- coding: utf-8 -*-
"""
ProgramEnrollment Views
"""
from __future__ import absolute_import, unicode_literals
import logging
from functools import wraps
from ccx_keys.locator import CCXLocator
from django.conf import settings
from django.core.exceptions import PermissionDenied
from django.core.management import call_command
from django.db import transaction
from django.http import Http404
from django.utils.functional import cached_property
from edx_rest_framework_extensions import permissions
from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
from edx_rest_framework_extensions.auth.session.authentication import SessionAuthenticationAllowInactiveUser
from opaque_keys.edx.keys import CourseKey
from organizations.models import Organization
from rest_framework import status
from rest_framework.exceptions import ValidationError
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from six import text_type
from course_modes.models import CourseMode
from lms.djangoapps.certificates.api import get_certificate_for_user
from lms.djangoapps.grades.api import CourseGradeFactory, clear_prefetched_course_grades, prefetch_course_grades
from lms.djangoapps.grades.rest_api.v1.utils import CourseEnrollmentPagination
from lms.djangoapps.program_enrollments.api.api import (
get_course_run_status,
get_course_run_url,
get_due_dates,
get_emails_enabled
)
from lms.djangoapps.program_enrollments.api.v1.constants import (
ENABLE_ENROLLMENT_RESET_FLAG,
MAX_ENROLLMENT_RECORDS,
CourseEnrollmentResponseStatuses,
ProgramEnrollmentResponseStatuses,
)
from lms.djangoapps.program_enrollments.api.v1.serializers import (
CourseRunOverviewListSerializer,
ProgramCourseEnrollmentListSerializer,
ProgramCourseEnrollmentRequestSerializer,
ProgramCourseGradeErrorResult,
ProgramCourseGradeResult,
ProgramCourseGradeResultSerializer,
ProgramEnrollmentCreateRequestSerializer,
ProgramEnrollmentListSerializer,
ProgramEnrollmentModifyRequestSerializer
)
from lms.djangoapps.program_enrollments.models import ProgramCourseEnrollment, ProgramEnrollment
from lms.djangoapps.program_enrollments.utils import (
ProviderDoesNotExistException,
get_provider_slug,
get_user_by_program_id
)
from openedx.core.djangoapps.catalog.utils import (
course_run_keys_for_program,
get_programs,
get_programs_by_type,
get_programs_for_organization,
normalize_program_type
)
from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
from openedx.core.lib.api.authentication import OAuth2AuthenticationAllowInactiveUser
from openedx.core.lib.api.view_utils import DeveloperErrorViewMixin, PaginatedAPIView, verify_course_exists
from student.helpers import get_resume_urls_for_enrollments
from student.models import CourseEnrollment
from student.roles import CourseInstructorRole, CourseStaffRole, UserBasedRole
from util.query import use_read_replica_if_available
logger = logging.getLogger(__name__)
def verify_program_exists(view_func):
"""
Raises:
An API error if the `program_uuid` kwarg in the wrapped function
does not exist in the catalog programs cache.
"""
@wraps(view_func)
def wrapped_function(self, request, **kwargs):
"""
Wraps the given view_function.
"""
program_uuid = kwargs['program_uuid']
program = get_programs(uuid=program_uuid)
if not program:
raise self.api_error(
status_code=status.HTTP_404_NOT_FOUND,
developer_message='no program exists with given key',
error_code='program_does_not_exist'
)
return view_func(self, request, **kwargs)
return wrapped_function
def verify_course_exists_and_in_program(view_func):
"""
Raises:
An api error if the course run specified by the `course_key` kwarg
in the wrapped function is not part of the curriculum of the program
specified by the `program_uuid` kwarg
Assumes that the program exists and that a program has exactly one active curriculum
"""
@wraps(view_func)
@verify_course_exists
def wrapped_function(self, request, **kwargs):
"""
Wraps view function
"""
course_key = CourseKey.from_string(kwargs['course_id'])
program_uuid = kwargs['program_uuid']
program = get_programs(uuid=program_uuid)
active_curricula = [c for c in program['curricula'] if c['is_active']]
if not active_curricula:
raise self.api_error(
status_code=status.HTTP_404_NOT_FOUND,
developer_message="the program does not have an active curriculum",
error_code='no_active_curriculum'
)
curriculum = active_curricula[0]
if not is_course_in_curriculum(curriculum, course_key):
raise self.api_error(
status_code=status.HTTP_404_NOT_FOUND,
developer_message="the program's curriculum does not contain the given course",
error_code='course_not_in_program'
)
return view_func(self, request, **kwargs)
def is_course_in_curriculum(curriculum, course_key):
for course in curriculum['courses']:
for course_run in course['course_runs']:
if CourseKey.from_string(course_run["key"]) == course_key:
return True
return wrapped_function
class ProgramEnrollmentPagination(CourseEnrollmentPagination):
"""
Pagination class for views in the Program Enrollments app.
"""
page_size = 100
class ProgramEnrollmentsView(DeveloperErrorViewMixin, PaginatedAPIView):
"""
A view for Create/Read/Update methods on Program Enrollment data.
Path: `/api/program_enrollments/v1/programs/{program_uuid}/enrollments/`
The path can contain an optional `page_size?=N` query parameter. The default page size is 100.
Returns:
* 200: OK - Contains a paginated set of program enrollment data.
* 401: The requesting user is not authenticated.
* 403: The requesting user lacks access for the given program.
* 404: The requested program does not exist.
Response:
In the case of a 200 response code, the response will include a paginated
data set. The `results` section of the response consists of a list of
program enrollment records, where each record contains the following keys:
* student_key: The identifier of the student enrolled in the program.
* status: The student's enrollment status.
* account_exists: A boolean indicating if the student has created an edx-platform user account.
* curriculum_uuid: The curriculum UUID of the enrollment record for the (student, program).
Example:
{
"next": null,
"previous": "http://testserver.com/api/program_enrollments/v1/programs/{program_uuid}/enrollments/?curor=abcd",
"results": [
{
"student_key": "user-0", "status": "pending",
"account_exists": False, "curriculum_uuid": "00000000-1111-2222-3333-444444444444"
},
{
"student_key": "user-1", "status": "pending",
"account_exists": False, "curriculum_uuid": "00000001-1111-2222-3333-444444444444"
},
{
"student_key": "user-2", "status": "enrolled",
"account_exists": True, "curriculum_uuid": "00000002-1111-2222-3333-444444444444"
},
{
"student_key": "user-3", "status": "enrolled",
"account_exists": True, "curriculum_uuid": "00000003-1111-2222-3333-444444444444"
},
],
}
Create
==========
Path: `/api/program_enrollments/v1/programs/{program_uuid}/enrollments/`
Where the program_uuid will be the uuid for a program.
Request body:
* The request body will be a list of one or more students to enroll with the following schema:
{
'status': A choice of the following statuses: ['enrolled', 'pending', 'canceled', 'suspended'],
student_key: string representation of a learner in partner systems,
'curriculum_uuid': string representation of a curriculum
}
Example:
[
{
"status": "enrolled",
"external_user_key": "123",
"curriculum_uuid": "2d7de549-b09e-4e50-835d-4c5c5080c566"
},{
"status": "canceled",
"external_user_key": "456",
"curriculum_uuid": "2d7de549-b09e-4e50-835d-4c5c5080c566"
},{
"status": "pending",
"external_user_key": "789",
"curriculum_uuid": "2d7de549-b09e-4e50-835d-4c5c5080c566"
},{
"status": "suspended",
"external_user_key": "abc",
"curriculum_uuid": "2d7de549-b09e-4e50-835d-4c5c5080c566"
},
]
Returns:
* Response Body: {<external_user_key>: <status>} with as many keys as there were in the request body
* external_user_key - string representation of a learner in partner systems
* status - the learner's registration status
* success statuses:
* 'enrolled'
* 'pending'
* 'canceled'
* 'suspended'
* failure statuses:
* 'duplicated' - the request body listed the same learner twice
* 'conflict' - there is an existing enrollment for that learner, curriculum and program combo
* 'invalid-status' - a status other than 'enrolled', 'pending', 'canceled', 'suspended' was entered
* 201: CREATED - All students were successfully enrolled.
* Example json response:
{
'123': 'enrolled',
'456': 'pending',
'789': 'canceled,
'abc': 'suspended'
}
* 207: MULTI-STATUS - Some students were successfully enrolled while others were not.
Details are included in the JSON response data.
* Example json response:
{
'123': 'duplicated',
'456': 'conflict',
'789': 'invalid-status,
'abc': 'suspended'
}
* 403: FORBIDDEN - The requesting user lacks access to enroll students in the given program.
* 404: NOT FOUND - The requested program does not exist.
* 413: PAYLOAD TOO LARGE - Over 25 students supplied
* 422: Unprocesable Entity - None of the students were successfully listed.
Update
==========
Path: `/api/program_enrollments/v1/programs/{program_uuid}/enrollments/`
Where the program_uuid will be the uuid for a program.
Request body:
* The request body will be a list of one or more students with their updated enrollment status:
{
'status': A choice of the following statuses: ['enrolled', 'pending', 'canceled', 'suspended'],
student_key: string representation of a learner in partner systems
}
Example:
[
{
"status": "enrolled",
"external_user_key": "123",
},{
"status": "canceled",
"external_user_key": "456",
},{
"status": "pending",
"external_user_key": "789",
},{
"status": "suspended",
"external_user_key": "abc",
},
]
Returns:
* Response Body: {<external_user_key>: <status>} with as many keys as there were in the request body
* external_user_key - string representation of a learner in partner systems
* status - the learner's registration status
* success statuses:
* 'enrolled'
* 'pending'
* 'canceled'
* 'suspended'
* failure statuses:
* 'duplicated' - the request body listed the same learner twice
* 'conflict' - there is an existing enrollment for that learner, curriculum and program combo
* 'invalid-status' - a status other than 'enrolled', 'pending', 'canceled', 'suspended' was entered
* 'not-in-program' - the user is not in the program and cannot be updated
* 201: CREATED - All students were successfully enrolled.
* Example json response:
{
'123': 'enrolled',
'456': 'pending',
'789': 'canceled,
'abc': 'suspended'
}
* 207: MULTI-STATUS - Some students were successfully enrolled while others were not.
Details are included in the JSON response data.
* Example json response:
{
'123': 'duplicated',
'456': 'not-in-program',
'789': 'invalid-status,
'abc': 'suspended'
}
* 403: FORBIDDEN - The requesting user lacks access to enroll students in the given program.
* 404: NOT FOUND - The requested program does not exist.
* 413: PAYLOAD TOO LARGE - Over 25 students supplied
* 422: Unprocesable Entity - None of the students were successfully updated.
"""
authentication_classes = (
JwtAuthentication,
OAuth2AuthenticationAllowInactiveUser,
SessionAuthenticationAllowInactiveUser,
)
permission_classes = (permissions.JWT_RESTRICTED_APPLICATION_OR_USER_ACCESS,)
pagination_class = ProgramEnrollmentPagination
@verify_program_exists
def get(self, request, program_uuid=None):
""" Defines the GET list endpoint for ProgramEnrollment objects. """
enrollments = use_read_replica_if_available(
ProgramEnrollment.objects.filter(program_uuid=program_uuid)
)
paginated_enrollments = self.paginate_queryset(enrollments)
serializer = ProgramEnrollmentListSerializer(paginated_enrollments, many=True)
return self.get_paginated_response(serializer.data)
@verify_program_exists
def post(self, request, *args, **kwargs):
"""
Create program enrollments for a list of learners
"""
return self.create_or_modify_enrollments(
request,
kwargs['program_uuid'],
ProgramEnrollmentCreateRequestSerializer,
self.create_program_enrollment,
status.HTTP_201_CREATED,
)
@verify_program_exists
def patch(self, request, **kwargs):
"""
Modify program enrollments for a list of learners
"""
return self.create_or_modify_enrollments(
request,
kwargs['program_uuid'],
ProgramEnrollmentModifyRequestSerializer,
self.modify_program_enrollment,
status.HTTP_200_OK,
)
@verify_program_exists
def put(self, request, **kwargs):
"""
Create/modify program enrollments for a list of learners
"""
return self.create_or_modify_enrollments(
request,
kwargs['program_uuid'],
ProgramEnrollmentCreateRequestSerializer,
self.create_or_modify_program_enrollment,
status.HTTP_200_OK,
)
def validate_enrollment_request(self, enrollment, seen_student_keys, serializer_class):
"""
Validates the given enrollment record and checks that it isn't a duplicate
"""
student_key = enrollment['student_key']
if student_key in seen_student_keys:
return CourseEnrollmentResponseStatuses.DUPLICATED
seen_student_keys.add(student_key)
enrollment_serializer = serializer_class(data=enrollment)
try:
enrollment_serializer.is_valid(raise_exception=True)
except ValidationError as e:
if enrollment_serializer.has_invalid_status():
return CourseEnrollmentResponseStatuses.INVALID_STATUS
else:
raise e
def create_or_modify_enrollments(self, request, program_uuid, serializer_class, operation, success_status):
"""
Process a list of program course enrollment request objects
and create or modify enrollments based on method
"""
results = {}
seen_student_keys = set()
enrollments = []
if not isinstance(request.data, list):
return Response('invalid enrollment record', status.HTTP_422_UNPROCESSABLE_ENTITY)
if len(request.data) > MAX_ENROLLMENT_RECORDS:
return Response(
'enrollment limit {}'.format(MAX_ENROLLMENT_RECORDS),
status.HTTP_413_REQUEST_ENTITY_TOO_LARGE
)
try:
for enrollment_request in request.data:
error_status = self.validate_enrollment_request(enrollment_request, seen_student_keys, serializer_class)
if error_status:
results[enrollment_request["student_key"]] = error_status
else:
enrollments.append(enrollment_request)
except KeyError: # student_key is not in enrollment_request
return Response('invalid enrollment record', status.HTTP_422_UNPROCESSABLE_ENTITY)
except TypeError: # enrollment_request isn't a dict
return Response('invalid enrollment record', status.HTTP_422_UNPROCESSABLE_ENTITY)
except ValidationError: # there was some other error raised by the serializer
return Response('invalid enrollment record', status.HTTP_422_UNPROCESSABLE_ENTITY)
program_enrollments = self.get_existing_program_enrollments(program_uuid, enrollments)
for enrollment in enrollments:
student_key = enrollment["student_key"]
if student_key in results and results[student_key] == ProgramEnrollmentResponseStatuses.DUPLICATED:
continue
try:
program_enrollment = program_enrollments[student_key]
except KeyError:
program_enrollment = None
results[student_key] = operation(enrollment, program_uuid, program_enrollment)
return self._get_created_or_updated_response(results, success_status)
def create_program_enrollment(self, request_data, program_uuid, program_enrollment):
"""
Create new ProgramEnrollment, unless the learner is already enrolled in the program
"""
if program_enrollment:
return ProgramEnrollmentResponseStatuses.CONFLICT
student_key = request_data.get('student_key')
try:
user = get_user_by_program_id(student_key, program_uuid)
except ProviderDoesNotExistException:
# IDP has not yet been set up, just create waiting enrollments
user = None
enrollment = ProgramEnrollment.objects.create(
user=user,
external_user_key=student_key,
program_uuid=program_uuid,
curriculum_uuid=request_data.get('curriculum_uuid'),
status=request_data.get('status')
)
return enrollment.status
# pylint: disable=unused-argument
def modify_program_enrollment(self, request_data, program_uuid, program_enrollment):
"""
Change the status of an existing program enrollment
"""
if not program_enrollment:
return ProgramEnrollmentResponseStatuses.NOT_IN_PROGRAM
program_enrollment.status = request_data.get('status')
program_enrollment.save()
return program_enrollment.status
def create_or_modify_program_enrollment(self, request_data, program_uuid, program_enrollment):
if program_enrollment:
return self.modify_program_enrollment(request_data, program_uuid, program_enrollment)
else:
return self.create_program_enrollment(request_data, program_uuid, program_enrollment)
def get_existing_program_enrollments(self, program_uuid, student_data):
""" Returns the existing program enrollments for the given students and program """
student_keys = [data['student_key'] for data in student_data]
return {
e.external_user_key: e
for e in ProgramEnrollment.bulk_read_by_student_key(program_uuid, student_keys)
}
def _get_created_or_updated_response(self, response_data, default_status=status.HTTP_201_CREATED):
"""
Helper method to determine an appropirate HTTP response status code.
"""
response_status = default_status
good_count = len([
v for v in response_data.values()
if v not in CourseEnrollmentResponseStatuses.ERROR_STATUSES
])
if not good_count:
response_status = status.HTTP_422_UNPROCESSABLE_ENTITY
elif good_count != len(response_data):
response_status = status.HTTP_207_MULTI_STATUS
return Response(
status=response_status,
data=response_data,
content_type='application/json',
)
class UserProgramReadOnlyAccessView(DeveloperErrorViewMixin, PaginatedAPIView):
"""
A view for checking the currently logged-in user's program read only access
There are three major categories of users this API is differentiating. See the table below.
--------------------------------------------------------------------------------------------
| User Type | API Returns |
--------------------------------------------------------------------------------------------
| edX staff | All programs |
--------------------------------------------------------------------------------------------
| course staff | All programs containing the courses of which the user is course staff |
--------------------------------------------------------------------------------------------
| learner | All programs the learner is enrolled in |
--------------------------------------------------------------------------------------------
Path: `/api/program_enrollments/v1/programs/enrollments/`
Returns:
* 200: OK - Contains a list of all programs in which the user has read only acccess to.
* 401: The requesting user is not authenticated.
The list will be a list of objects with the following keys:
* `uuid` - the identifier of the program in which the user has read only access to.
* `slug` - the string from which a link to the corresponding program page can be constructed.
Example:
[
{
'uuid': '00000000-1111-2222-3333-444444444444',
'slug': 'deadbeef'
},
{
'uuid': '00000000-1111-2222-3333-444444444445',
'slug': 'undead-cattle'
}
]
"""
authentication_classes = (
JwtAuthentication,
OAuth2AuthenticationAllowInactiveUser,
SessionAuthenticationAllowInactiveUser,
)
permission_classes = (IsAuthenticated,)
DEFAULT_PROGRAM_TYPE = 'masters'
def get(self, request):
"""
How to respond to a GET request to this endpoint
"""
request_user = request.user
programs = []
requested_program_type = normalize_program_type(request.GET.get('type', self.DEFAULT_PROGRAM_TYPE))
if request_user.is_staff:
programs = get_programs_by_type(request.site, requested_program_type)
elif self.is_course_staff(request_user):
programs = self.get_programs_user_is_course_staff_for(request_user, requested_program_type)
else:
program_enrollments = ProgramEnrollment.objects.filter(
user=request.user,
status__in=('enrolled', 'pending')
)
uuids = [enrollment.program_uuid for enrollment in program_enrollments]
programs = get_programs(uuids=uuids) or []
programs_in_which_user_has_access = [
{'uuid': program['uuid'], 'slug': program['marketing_slug']}
for program in programs
]
return Response(programs_in_which_user_has_access, status.HTTP_200_OK)
def is_course_staff(self, user):
"""
Returns true if the user is a course_staff member of any course within a program
"""
staff_course_keys = self.get_course_keys_user_is_staff_for(user)
return len(staff_course_keys)
def get_course_keys_user_is_staff_for(self, user):
"""
Return all the course keys the user is course instructor or course staff role for
"""
# Get all the courses of which the user is course staff for. If None, return false
def filter_ccx(course_access):
""" CCXs cannot be edited in Studio and should not be filtered """
return not isinstance(course_access.course_id, CCXLocator)
instructor_courses = UserBasedRole(user, CourseInstructorRole.ROLE).courses_with_role()
staff_courses = UserBasedRole(user, CourseStaffRole.ROLE).courses_with_role()
all_courses = list(filter(filter_ccx, instructor_courses | staff_courses))
course_keys = {}
for course_access in all_courses:
if course_access.course_id is not None:
course_keys[course_access.course_id] = course_access.course_id
return list(course_keys.values())
def get_programs_user_is_course_staff_for(self, user, program_type_filter):
"""
Return a list of programs the user is course staff for.
This function would take a list of course runs the user is staff of, and then
try to get the Masters program associated with each course_runs.
"""
program_list = []
for course_key in self.get_course_keys_user_is_staff_for(user):
course_run_programs = get_programs(course=course_key)
for course_run_program in course_run_programs:
if course_run_program and course_run_program.get('type').lower() == program_type_filter:
program_list.append(course_run_program)
return program_list
class ProgramSpecificViewMixin(object):
"""
A mixin for views that operate on or within a specific program.
"""
@cached_property
def program(self):
"""
The program specified by the `program_uuid` URL parameter.
"""
program = get_programs(uuid=self.kwargs['program_uuid'])
if program is None:
raise Http404()
return program
class ProgramCourseRunSpecificViewMixin(ProgramSpecificViewMixin):
"""
A mixin for views that operate on or within a specific course run in a program
"""
@property
def course_key(self):
"""
The course key for the course run specified by the `course_id` URL parameter.
"""
return CourseKey.from_string(self.kwargs['course_id'])
# pylint: disable=line-too-long
class ProgramCourseEnrollmentsView(DeveloperErrorViewMixin, ProgramCourseRunSpecificViewMixin, PaginatedAPIView):
"""
A view for enrolling students in a course through a program,
modifying program course enrollments, and listing program course
enrollments.
Path: ``/api/program_enrollments/v1/programs/{program_uuid}/courses/{course_id}/enrollments/``
Accepts: [GET, POST, PATCH, PUT]
For GET requests, the path can contain an optional `page_size?=N` query parameter.
The default page size is 100.
------------------------------------------------------------------------------------
POST, PATCH, PUT
------------------------------------------------------------------------------------
**Returns**
* 200: Returns a map of students and their enrollment status.
* 207: Not all students enrolled. Returns resulting enrollment status.
* 401: User is not authenticated
* 403: User lacks read access organization of specified program.
* 404: Program does not exist, or course does not exist in program
* 422: Invalid request, unable to enroll students.
------------------------------------------------------------------------------------
GET
------------------------------------------------------------------------------------
**Returns**
* 200: OK - Contains a paginated set of program course enrollment data.
* 401: The requesting user is not authenticated.
* 403: The requesting user lacks access for the given program/course.
* 404: The requested program or course does not exist.
**Response**
In the case of a 200 response code, the response will include a paginated
data set. The `results` section of the response consists of a list of
program course enrollment records, where each record contains the following keys:
* student_key: The identifier of the student enrolled in the program and course.
* status: The student's course enrollment status.
* account_exists: A boolean indicating if the student has created an edx-platform user account.
* curriculum_uuid: The curriculum UUID of the enrollment record for the (student, program).
**Example**
{
"next": null,
"previous": "http://testserver.com/api/program_enrollments/v1/programs/{program_uuid}/courses/{course_id}/enrollments/?curor=abcd",
"results": [
{
"student_key": "user-0", "status": "inactive",
"account_exists": False, "curriculum_uuid": "00000000-1111-2222-3333-444444444444"
},
{
"student_key": "user-1", "status": "inactive",
"account_exists": False, "curriculum_uuid": "00000001-1111-2222-3333-444444444444"
},
{
"student_key": "user-2", "status": "active",
"account_exists": True, "curriculum_uuid": "00000002-1111-2222-3333-444444444444"
},
{
"student_key": "user-3", "status": "active",
"account_exists": True, "curriculum_uuid": "00000003-1111-2222-3333-444444444444"
},
],
}
"""
authentication_classes = (
JwtAuthentication,
OAuth2AuthenticationAllowInactiveUser,
SessionAuthenticationAllowInactiveUser,
)
permission_classes = (permissions.JWT_RESTRICTED_APPLICATION_OR_USER_ACCESS,)
pagination_class = ProgramEnrollmentPagination
@verify_course_exists
@verify_program_exists
def get(self, request, program_uuid=None, course_id=None):
""" Defines the GET list endpoint for ProgramCourseEnrollment objects. """
course_key = CourseKey.from_string(course_id)
enrollments = use_read_replica_if_available(
ProgramCourseEnrollment.objects.filter(
program_enrollment__program_uuid=program_uuid, course_key=course_key
).select_related(
'program_enrollment'
)
)
paginated_enrollments = self.paginate_queryset(enrollments)
serializer = ProgramCourseEnrollmentListSerializer(paginated_enrollments, many=True)
return self.get_paginated_response(serializer.data)
@verify_program_exists
@verify_course_exists_and_in_program
def post(self, request, program_uuid=None, course_id=None):
"""
Enroll a list of students in a course in a program
"""
return self.create_or_modify_enrollments(
request,
program_uuid,
self.enroll_learner_in_course
)
@verify_program_exists
@verify_course_exists_and_in_program
# pylint: disable=unused-argument
def patch(self, request, program_uuid=None, course_id=None):
"""
Modify the program course enrollments of a list of learners
"""
return self.create_or_modify_enrollments(
request,
program_uuid,
self.modify_learner_enrollment_status
)
@verify_program_exists
@verify_course_exists_and_in_program
# pylint: disable=unused-argument
def put(self, request, program_uuid=None, course_id=None):
"""
Create or Update the program course enrollments of a list of learners
"""
return self.create_or_modify_enrollments(
request,
program_uuid,
self.create_or_update_learner_enrollment
)
def create_or_modify_enrollments(self, request, program_uuid, operation):
"""
Process a list of program course enrollment request objects
and create or modify enrollments based on method
"""
results = {}
seen_student_keys = set()
enrollments = []
if not isinstance(request.data, list):
return Response('invalid enrollment record', status.HTTP_400_BAD_REQUEST)
if len(request.data) > MAX_ENROLLMENT_RECORDS:
return Response(
'enrollment limit 25', status.HTTP_413_REQUEST_ENTITY_TOO_LARGE
)
try:
for enrollment_request in request.data:
error_status = self.check_enrollment_request(enrollment_request, seen_student_keys)
if error_status:
results[enrollment_request["student_key"]] = error_status
else:
enrollments.append(enrollment_request)
except KeyError: # student_key is not in enrollment_request
return Response('invalid enrollment record', status.HTTP_400_BAD_REQUEST)
except TypeError: # enrollment_request isn't a dict
return Response('invalid enrollment record', status.HTTP_400_BAD_REQUEST)
except ValidationError: # there was some other error raised by the serializer
return Response('invalid enrollment record', status.HTTP_400_BAD_REQUEST)
program_enrollments = self.get_existing_program_enrollments(program_uuid, enrollments)
for enrollment in enrollments:
student_key = enrollment["student_key"]
if student_key in results and results[student_key] == CourseEnrollmentResponseStatuses.DUPLICATED:
continue
try:
program_enrollment = program_enrollments[student_key]
except KeyError:
results[student_key] = CourseEnrollmentResponseStatuses.NOT_IN_PROGRAM
else:
program_course_enrollment = program_enrollment.get_program_course_enrollment(self.course_key)
results[student_key] = operation(enrollment, program_enrollment, program_course_enrollment)
good_count = sum(1 for _, v in results.items() if v not in CourseEnrollmentResponseStatuses.ERROR_STATUSES)
if not good_count:
return Response(results, status.HTTP_422_UNPROCESSABLE_ENTITY)
if good_count != len(results):
return Response(results, status.HTTP_207_MULTI_STATUS)
else:
return Response(results)
def check_enrollment_request(self, enrollment, seen_student_keys):
"""
Checks that the given enrollment record is valid and hasn't been duplicated
"""
student_key = enrollment['student_key']
if student_key in seen_student_keys:
return CourseEnrollmentResponseStatuses.DUPLICATED
seen_student_keys.add(student_key)
enrollment_serializer = ProgramCourseEnrollmentRequestSerializer(data=enrollment)
try:
enrollment_serializer.is_valid(raise_exception=True)
except ValidationError as e:
if enrollment_serializer.has_invalid_status():
return CourseEnrollmentResponseStatuses.INVALID_STATUS
else:
raise e
def get_existing_program_enrollments(self, program_uuid, enrollments):
"""
Parameters:
- enrollments: A list of enrollment requests
Returns:
- Dictionary mapping all student keys in the enrollment requests
to that user's existing program enrollment in <self.program>
"""
external_user_keys = [e["student_key"] for e in enrollments]
existing_enrollments = ProgramEnrollment.objects.filter(
external_user_key__in=external_user_keys,
program_uuid=program_uuid,
)
existing_enrollments = existing_enrollments.prefetch_related('program_course_enrollments')
return {enrollment.external_user_key: enrollment for enrollment in existing_enrollments}
def enroll_learner_in_course(self, enrollment_request, program_enrollment, program_course_enrollment):
"""
Attempts to enroll the specified user into the course as a part of the
given program enrollment with the given status
Returns the actual status
"""
if program_course_enrollment:
return CourseEnrollmentResponseStatuses.CONFLICT
return ProgramCourseEnrollment.create_program_course_enrollment(
program_enrollment,
self.course_key,
enrollment_request['status']
)
# pylint: disable=unused-argument
def modify_learner_enrollment_status(self, enrollment_request, program_enrollment, program_course_enrollment):
"""
Attempts to modify the specified user's enrollment in the given course
in the given program
"""
if program_course_enrollment is None:
return CourseEnrollmentResponseStatuses.NOT_FOUND
return program_course_enrollment.change_status(enrollment_request['status'])
def create_or_update_learner_enrollment(self, enrollment_request, program_enrollment, program_course_enrollment):
"""
Attempts to create or update the specified user's enrollment in the given course
in the given program
"""
if program_course_enrollment is None:
# create the course enrollment
return ProgramCourseEnrollment.create_program_course_enrollment(
program_enrollment,
self.course_key,
enrollment_request['status']
)
else:
# Update course enrollment
return program_course_enrollment.change_status(enrollment_request['status'])
class ProgramCourseEnrollmentOverviewView(DeveloperErrorViewMixin, ProgramSpecificViewMixin, APIView):
"""
A view for getting data associated with a user's course enrollments
as part of a program enrollment.
Path: ``/api/program_enrollments/v1/programs/{program_uuid}/overview/``
Accepts: [GET]
------------------------------------------------------------------------------------
GET
------------------------------------------------------------------------------------
**Returns**
* 200: OK - Contains an object of user program course enrollment data.
* 401: Unauthorized - The requesting user is not authenticated.
* 403: Forbidden -The requesting user lacks access for the given program.
* 404: Not Found - The requested program does not exist.
**Response**
In the case of a 200 response code, the response will include a
data set. The `course_runs` section of the response consists of a list of
program course enrollment overview, where each overview contains the following keys:
* course_run_id: the id for the course run
* display_name: display name of the course run
* resume_course_run_url: the absolute url that takes the user back to
their position in the course run;
if absent, user has not made progress in the course
* course_run_url: the absolute url for the course run
* start_date: the start date for the course run; null if no start date
* end_date: the end date for the course run' null if no end date
* course_run_status: the status of the course; one of "in_progress", "upcoming", and "completed"
* emails_enabled: boolean representing whether emails are enabled for the course;
if absent, the bulk email feature is either not enable at the platform level or is not enabled for the course;
if True or False, bulk email feature is enabled, and value represents whether or not user wants to receive emails
* due_dates: a list of subsection due dates for the course run:
** name: name of the subsection
** url: deep link to the subsection
** date: due date for the subsection
* micromasters_title: title of the MicroMasters program that the course run is a part of;
if absent, the course run is not a part of a MicroMasters program
* certificate_download_url: url to download a certificate, if available;
if absent, certificate is not downloadable
**Example**
{
"course_runs": [
{
"course_run_id": "edX+AnimalsX+Aardvarks",
"display_name": "Astonishing Aardvarks",
"course_run_url": "https://courses.edx.org/courses/course-v1:edX+AnimalsX+Aardvarks/course/",
"start_date": "2017-02-05T05:00:00Z",
"end_date": "2018-02-05T05:00:00Z",
"course_run_status": "completed"
"emails_enabled": true,
"due_dates": [
{
"name": "Introduction: What even is an aardvark?",
"url": "https://courses.edx.org/courses/course-v1:edX+AnimalsX+Aardvarks/jump_to/block-v1:edX+AnimalsX+Aardvarks+type@chapter+block@1414ffd5143b4b508f739b563ab468b7",
"date": "2017-05-01T05:00:00Z"
},
{
"name": "Quiz: Aardvark or Anteater?",
"url": "https://courses.edx.org/courses/course-v1:edX+AnimalsX+Aardvarks/jump_to/block-v1:edX+AnimalsX+Aardvarks+type@sequential+block@edx_introduction",
"date": "2017-03-05T00:00:00Z"
}
],
"micromasters_title": "Animals",
"certificate_download_url": "https://courses.edx.org/certificates/123"
},
{
"course_run_id": "edX+AnimalsX+Baboons",
"display_name": "Breathtaking Baboons",
"course_run_url": "https://courses.edx.org/courses/course-v1:edX+AnimalsX+Baboons/course/",
"start_date": "2018-02-05T05:00:00Z",
"end_date": null,
"course_run_status": "in_progress"
"emails_enabled": false,
"due_dates": [],
"micromasters_title": "Animals",
"certificate_download_url": "https://courses.edx.org/certificates/123",
"resume_course_run_url": "https://courses.edx.org/courses/course-v1:edX+AnimalsX+Baboons/jump_to/block-v1:edX+AnimalsX+Baboons+type@sequential+block@edx_introduction"
}
]
}
"""
authentication_classes = (
JwtAuthentication,
OAuth2AuthenticationAllowInactiveUser,
SessionAuthenticationAllowInactiveUser,
)
permission_classes = (IsAuthenticated,)
@verify_program_exists
def get(self, request, program_uuid=None):
"""
Defines the GET endpoint for overviews of course enrollments
for a user as part of a program.
"""
user = request.user
self._check_program_enrollment_exists(user, program_uuid)
program = get_programs(uuid=program_uuid)
course_run_keys = [CourseKey.from_string(key) for key in course_run_keys_for_program(program)]
course_enrollments = CourseEnrollment.objects.filter(
user=user,
course_id__in=course_run_keys,
mode__in=[CourseMode.VERIFIED, CourseMode.MASTERS],
is_active=True,
)
overviews = CourseOverview.get_from_ids_if_exists(course_run_keys)
course_run_resume_urls = get_resume_urls_for_enrollments(user, course_enrollments)
course_runs = []
for enrollment in course_enrollments:
overview = overviews[enrollment.course_id]
certificate_info = get_certificate_for_user(user.username, enrollment.course_id) or {}
course_run_dict = {
'course_run_id': enrollment.course_id,
'display_name': overview.display_name_with_default,
'course_run_status': get_course_run_status(overview, certificate_info),
'course_run_url': get_course_run_url(request, enrollment.course_id),
'start_date': overview.start,
'end_date': overview.end,
'due_dates': get_due_dates(request, enrollment.course_id, user),
}
emails_enabled = get_emails_enabled(user, enrollment.course_id)
if emails_enabled is not None:
course_run_dict['emails_enabled'] = emails_enabled
if certificate_info.get('download_url'):
course_run_dict['certificate_download_url'] = request.build_absolute_uri(
certificate_info['download_url']
)
if self.program['type'] == 'MicroMasters':
course_run_dict['micromasters_title'] = self.program['title']
if course_run_resume_urls.get(enrollment.course_id):
relative_resume_course_run_url = course_run_resume_urls.get(
enrollment.course_id
)
course_run_dict['resume_course_run_url'] = request.build_absolute_uri(
relative_resume_course_run_url
)
course_runs.append(course_run_dict)
serializer = CourseRunOverviewListSerializer({'course_runs': course_runs})
return Response(serializer.data)
@staticmethod
def _check_program_enrollment_exists(user, program_uuid):
"""
Raises ``PermissionDenied`` if the user is not enrolled in the program with the given UUID.
"""
program_enrollments = ProgramEnrollment.objects.filter(
program_uuid=program_uuid,
user=user,
status='enrolled',
)
if not program_enrollments:
raise PermissionDenied
class ProgramCourseGradesView(
DeveloperErrorViewMixin,
ProgramCourseRunSpecificViewMixin,
PaginatedAPIView,
):
"""
A view for retrieving a paginated list of grades for all students enrolled
in a given courserun through a given program.
Path: ``/api/program_enrollments/v1/programs/{program_uuid}/courses/{course_id}/grades/``
Accepts: [GET]
For GET requests, the path can contain an optional `page_size?=N` query parameter.
The default page size is 100.
------------------------------------------------------------------------------------
GETs
------------------------------------------------------------------------------------
**Returns**
* 200: OK - Contains a paginated set of program courserun grades.
* 204: No Content - No grades to return
* 207: Mixed result - Contains mixed list of program courserun grades
and grade-fetching errors
* 422: All failed - Contains list of grade-fetching errors
* 401: The requesting user is not authenticated.
* 403: The requesting user lacks access for the given program/course.
* 404: The requested program or course does not exist.
**Response**
In the case of a 200/207/422 response code, the response will include a
paginated data set. The `results` section of the response consists of a
list of grade records, where each successfully loaded record contains:
* student_key: The identifier of the student enrolled in the program and course.
* letter_grade: A letter grade as defined in grading policy
(e.g. 'A' 'B' 'C' for 6.002x) or None.
* passed: Boolean representing whether the course has been
passed according to the course's grading policy.
* percent: A float representing the overall grade for the course.
and failed-to-load records contain:
* student_key
* error: error message from grades Exception
**Example**
207 Multi-Status
{
"next": null,
"previous": "http://example.com/api/program_enrollments/v1/programs/{program_uuid}/courses/{course_id}/grades/?cursor=abcd",
"results": [;
{
"student_key": "01709bffeae2807b6a7317",
"letter_grade": "Pass",
"percent": 0.95,
"passed": true
},
{
"student_key": "2cfe15e3380a52e7198237",
"error": "Timeout while calculating grade"
},
...
],
}
"""
authentication_classes = (
JwtAuthentication,
OAuth2AuthenticationAllowInactiveUser,
SessionAuthenticationAllowInactiveUser,
)
permission_classes = (permissions.JWT_RESTRICTED_APPLICATION_OR_USER_ACCESS,)
pagination_class = ProgramEnrollmentPagination
@verify_course_exists
@verify_program_exists
def get(self, request, program_uuid=None, course_id=None):
"""
Defines the GET list endpoint for ProgramCourseGrade objects.
"""
course_key = CourseKey.from_string(course_id)
grade_results = self._load_grade_results(program_uuid, course_key)
serializer = ProgramCourseGradeResultSerializer(grade_results, many=True)
response_code = self._calc_response_code(grade_results)
return self.get_paginated_response(serializer.data, status_code=response_code)
def _load_grade_results(self, program_uuid, course_key):
"""
Load grades (or grading errors) for a given program courserun.
Arguments:
program_uuid (str)
course_key (CourseKey)
Returns: list[ProgramCourseGradeResult|ProgramCourseGradeErrorResult]
"""
enrollments_qs = use_read_replica_if_available(
ProgramCourseEnrollment.objects.filter(
program_enrollment__program_uuid=program_uuid,
program_enrollment__user__isnull=False,
course_key=course_key,
).select_related(
'program_enrollment',
'program_enrollment__user',
)
)
paginated_enrollments = self.paginate_queryset(enrollments_qs)
if not paginated_enrollments:
return []
# Hint: `zip(*(list))` can be read as "unzip(list)"
enrollments, users = zip(*(
(enrollment, enrollment.program_enrollment.user)
for enrollment in paginated_enrollments
))
enrollment_grade_pairs = zip(
enrollments, self._iter_grades(course_key, list(users))
)
grade_results = [
(
ProgramCourseGradeResult(enrollment, grade)
if grade
else ProgramCourseGradeErrorResult(enrollment, exception)
)
for enrollment, (grade, exception) in enrollment_grade_pairs
]
return grade_results
@staticmethod
def _iter_grades(course_key, users):
"""
Load a user grades for a course, using bulk fetching for efficiency.
Arguments:
course_key (CourseKey)
users (list[User])
Returns: iterable[( CourseGradeBase|NoneType, Exception|NoneType )]
Iterable of pairs, in same order as `users`.
The first item in the pair is the grade, or None if loading the
grade failed.
The second item in the pair is an exception or None.
"""
prefetch_course_grades(course_key, users)
try:
grades_iter = CourseGradeFactory().iter(users, course_key=course_key)
for user, course_grade, exception in grades_iter:
if not course_grade:
fmt = 'Failed to load course grade for user ID {} in {}: {}'
err_str = fmt.format(
user.id,
course_key,
text_type(exception) if exception else 'Unknown error'
)
logger.error(err_str)
yield course_grade, exception
finally:
clear_prefetched_course_grades(course_key)
@staticmethod
def _calc_response_code(grade_results):
"""
Returns HTTP status code appropriate for list of results,
which may be grades or errors.
Arguments:
enrollment_grade_results: list[ProgramCourseGradeResult]
Returns: int
* 200 for all success
* 207 for mixed result
* 422 for all failure
* 204 for empty
"""
if not grade_results:
return status.HTTP_204_NO_CONTENT
if all(result.is_error for result in grade_results):
return status.HTTP_422_UNPROCESSABLE_ENTITY
if any(result.is_error for result in grade_results):
return status.HTTP_207_MULTI_STATUS
return status.HTTP_200_OK
class EnrollmentDataResetView(APIView):
"""
Resets enrollments and users for a given organization and set of programs.
Note, this will remove ALL users from the input organization.
Path: ``/api/program_enrollments/v1/integration-reset/``
Accepts: [POST]
------------------------------------------------------------------------------------
POST
------------------------------------------------------------------------------------
**Returns**
* 200: OK - Enrollments and users sucessfully deleted
* 400: Bad Requeset - Program does not match the requested organization
* 401: Unauthorized - The requesting user is not authenticated.
* 404: Not Found - A requested program does not exist.
**Response**
"""
authentication_classes = (
JwtAuthentication,
OAuth2AuthenticationAllowInactiveUser,
SessionAuthenticationAllowInactiveUser,
)
permission_classes = (permissions.JWT_RESTRICTED_APPLICATION_OR_USER_ACCESS,)
@transaction.atomic
def post(self, request):
"""
Reset enrollment and user data for organization
"""
if not settings.FEATURES.get(ENABLE_ENROLLMENT_RESET_FLAG):
return Response('reset not enabled on this environment', status.HTTP_501_NOT_IMPLEMENTED)
try:
org_key = request.data['organization']
except KeyError:
return Response("missing required body content 'organization'", status.HTTP_400_BAD_REQUEST)
try:
organization = Organization.objects.get(short_name=org_key)
except Organization.DoesNotExist:
return Response('organization {} not found'.format(org_key), status.HTTP_404_NOT_FOUND)
try:
idp_slug = get_provider_slug(organization)
call_command('remove_social_auth_users', idp_slug, force=True)
except ProviderDoesNotExistException:
pass
programs = get_programs_for_organization(organization=organization.short_name)
if programs:
call_command('reset_enrollment_data', ','.join(programs), force=True)
return Response('success')