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

add program course enrollment PATCH

Browse Source
This commit is contained in:
jansenk
2019-05-03 16:07:43 -04:00
committed by Jansen Kantor
parent 6363090144
commit 1caa647621
5 changed files with 330 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -18,6 +18,7 @@ class CourseEnrollmentResponseStatuses(object):
CONFLICT = "conflict"
ILLEGAL_OPERATION = "illegal-operation"
NOT_IN_PROGRAM = "not-in-program"
NOT_FOUND = "not-found"
INTERNAL_ERROR = "internal-error"
ERROR_STATUSES = (
@@ -26,5 +27,6 @@ class CourseEnrollmentResponseStatuses(object):
CONFLICT,
ILLEGAL_OPERATION,
NOT_IN_PROGRAM,
NOT_FOUND,
INTERNAL_ERROR,
)

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

@@ -12,14 +12,14 @@ from django.urls import reverse
from django.contrib.auth.models import User
import mock
from opaque_keys.edx.keys import CourseKey
from rest_framework import status
from rest_framework.test import APITestCase
from six import text_type
from lms.djangoapps.courseware.tests.factories import GlobalStaffFactory
from lms.djangoapps.program_enrollments.api.v1.constants import CourseEnrollmentResponseStatuses as CourseStatuses
from lms.djangoapps.program_enrollments.models import ProgramCourseEnrollment, ProgramEnrollment
from lms.djangoapps.program_enrollments.models import ProgramEnrollment, ProgramCourseEnrollment
from student.tests.factories import UserFactory, CourseEnrollmentFactory
from openedx.core.djangoapps.catalog.cache import PROGRAM_CACHE_KEY_TPL
from openedx.core.djangoapps.catalog.tests.factories import (
CourseFactory,
@@ -28,8 +28,6 @@ from openedx.core.djangoapps.catalog.tests.factories import (
)
from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
from openedx.core.djangolib.testing.utils import CacheIsolationMixin
from student.tests.factories import UserFactory
from .factories import ProgramCourseEnrollmentFactory, ProgramEnrollmentFactory
@@ -229,12 +227,15 @@ class ProgramCacheTestCaseMixin(CacheIsolationMixin):
@ddt.ddt
class CourseEnrollmentPostTests(APITestCase, ProgramCacheTestCaseMixin):
""" Tests for mock course enrollment """
class BaseCourseEnrollmentTestsMixin(ProgramCacheTestCaseMixin):
"""
A base for tests for course enrollment.
Children should override self.request()
"""
@classmethod
def setUpClass(cls):
super(CourseEnrollmentPostTests, cls).setUpClass()
super(BaseCourseEnrollmentTestsMixin, cls).setUpClass()
cls.start_cache_isolation()
cls.password = 'password'
cls.student = UserFactory.create(username='student', password=cls.password)
@@ -243,10 +244,10 @@ class CourseEnrollmentPostTests(APITestCase, ProgramCacheTestCaseMixin):
@classmethod
def tearDownClass(cls):
cls.end_cache_isolation()
super(CourseEnrollmentPostTests, cls).tearDownClass()
super(BaseCourseEnrollmentTestsMixin, cls).tearDownClass()
def setUp(self):
super(CourseEnrollmentPostTests, self).setUp()
super(BaseCourseEnrollmentTestsMixin, self).setUp()
self.clear_caches()
self.addCleanup(self.clear_caches)
self.program_uuid = uuid4()
@@ -282,6 +283,9 @@ class CourseEnrollmentPostTests(APITestCase, ProgramCacheTestCaseMixin):
}
)
def request(self, path, data):
pass
def create_program_enrollment(self, external_user_key, user=False):
"""
Creates and returns a ProgramEnrollment for the given external_user_key and
@@ -296,37 +300,34 @@ class CourseEnrollmentPostTests(APITestCase, ProgramCacheTestCaseMixin):
program_enrollment.save()
return program_enrollment
def test_enrollments(self):
self.create_program_enrollment('l1')
self.create_program_enrollment('l2')
self.create_program_enrollment('l3', user=None)
self.create_program_enrollment('l4', user=None)
post_data = [
self.learner_enrollment("l1", "active"),
self.learner_enrollment("l2", "inactive"),
self.learner_enrollment("l3", "active"),
self.learner_enrollment("l4", "inactive"),
]
response = self.client.post(self.default_url, post_data, format="json")
self.assertEqual(200, response.status_code)
self.assertDictEqual(
{
"l1": "active",
"l2": "inactive",
"l3": "active",
"l4": "inactive",
},
response.data
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,
)
course_enrollment.is_active = course_status == "active"
course_enrollment.save()
return ProgramCourseEnrollmentFactory(
program_enrollment=program_enrollment,
course_key=self.course_key,
course_enrollment=course_enrollment,
status=course_status,
)
self.assert_program_course_enrollment("l1", "active", True)
self.assert_program_course_enrollment("l2", "inactive", True)
self.assert_program_course_enrollment("l3", "active", False)
self.assert_program_course_enrollment("l4", "inactive", False)
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)
def assert_program_course_enrollment(self, external_user_key, expected_status, has_user):
"""
Convenience method to assert that a ProgramCourseEnrollment has been created,
and potentially that a CourseEnrollment has also been created
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,
@@ -342,81 +343,68 @@ class CourseEnrollmentPostTests(APITestCase, ProgramCacheTestCaseMixin):
else:
self.assertIsNone(course_enrollment)
def test_duplicate(self):
post_data = [
self.learner_enrollment("l1", "active"),
self.learner_enrollment("l1", "active"),
]
response = self.client.post(self.default_url, post_data, format="json")
self.assertEqual(422, response.status_code)
self.assertDictEqual(
{
"l1": CourseStatuses.DUPLICATED
},
response.data
)
def test_conflict(self):
program_enrollment = self.create_program_enrollment('l1')
ProgramCourseEnrollmentFactory.create(
program_enrollment=program_enrollment,
course_key=self.course_key
)
post_data = [self.learner_enrollment("l1")]
response = self.client.post(self.default_url, post_data, format="json")
self.assertEqual(422, response.status_code)
self.assertDictEqual({'l1': CourseStatuses.CONFLICT}, response.data)
def test_user_not_in_program(self):
self.create_program_enrollment('l1')
post_data = [
self.learner_enrollment("l1"),
self.learner_enrollment("l2"),
]
response = self.client.post(self.default_url, post_data, format="json")
self.assertEqual(207, response.status_code)
self.assertDictEqual(
{
"l1": "active",
"l2": "not-in-program",
},
response.data
)
def test_401_not_logged_in(self):
self.client.logout()
post_data = [self.learner_enrollment("A")]
response = self.client.post(self.default_url, post_data, format="json")
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.client.login(username=self.student, password=self.password)
post_data = [self.learner_enrollment("A")]
response = self.client.post(self.default_url, post_data, format="json")
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):
post_data = [self.learner_enrollment(str(i)) for i in range(30)]
response = self.client.post(self.default_url, post_data, format="json")
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_program(self):
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),
self.get_url(self.program_uuid, CourseKey.from_string("course-v1:fake+fake+fake")),
self.get_url(self.program_uuid, self.course_not_in_program_key),
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
]
post_data = [self.learner_enrollment("A")]
request_data = [self.learner_enrollment("learner-1")]
for path_404 in paths:
response = self.client.post(path_404, post_data, format="json")
response = self.request(path_404, request_data)
self.assertEqual(404, response.status_code)
def test_invalid_status(self):
post_data = [self.learner_enrollment('A', 'this-is-not-a-status')]
response = self.client.post(self.default_url, post_data, format="json")
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({'A': CourseStatuses.INVALID_STATUS}, response.data)
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'}],
@@ -425,12 +413,121 @@ class CourseEnrollmentPostTests(APITestCase, ProgramCacheTestCaseMixin):
[{'student_key': '000', 'status': 'active'}, "bad_data"],
"not a list",
)
def test_422_unprocessable_entity_bad_data(self, post_data):
response = self.client.post(self.default_url, post_data, format="json")
def test_422_unprocessable_entity_bad_data(self, request_data):
response = self.request(self.default_url, request_data)
self.assertEqual(response.status_code, 422)
self.assertIn('invalid enrollment record', response.data)
class CourseEnrollmentPostTests(BaseCourseEnrollmentTestsMixin, APITestCase):
""" Tests for course enrollment POST """
def request(self, path, data):
return self.client.post(path, data, format='json')
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_user_already_enrolled_in_course(self):
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_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 CourseEnrollmentPatchTests(BaseCourseEnrollmentTestsMixin, APITestCase):
""" Tests for course enrollment PATCH """
def request(self, path, data):
return self.client.patch(path, data, format='json')
def test_207_multistatus(self):
self.create_program_and_course_enrollments('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
)
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.assertEqual(422, response.status_code)
self.assertDictEqual({'learner-1': CourseStatuses.NOT_FOUND}, response.data)
@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.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)
patch_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, patch_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 ProgramCourseEnrollmentListTest(ListViewTestMixin, APITestCase):
"""
Tests for GET calls to the Program Course Enrollments API.

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

@@ -449,6 +449,28 @@ class ProgramCourseEnrollmentsView(DeveloperErrorViewMixin, ProgramCourseRunSpec
"""
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
)
# 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
)
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
"""
self.check_course_existence_and_membership()
results = {}
seen_student_keys = set()
@@ -480,7 +502,13 @@ class ProgramCourseEnrollmentsView(DeveloperErrorViewMixin, ProgramCourseRunSpec
student_key = enrollment["student_key"]
if student_key in results and results[student_key] == CourseEnrollmentResponseStatuses.DUPLICATED:
continue
results[student_key] = self.enroll_learner_in_course(enrollment, program_enrollments)
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:
@@ -523,24 +551,27 @@ class ProgramCourseEnrollmentsView(DeveloperErrorViewMixin, ProgramCourseRunSpec
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_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
"""
student_key = enrollment_request['student_key']
try:
program_enrollment = program_enrollments[student_key]
except KeyError:
return CourseEnrollmentResponseStatuses.NOT_IN_PROGRAM
if program_enrollment.get_program_course_enrollment(self.course_key):
if program_course_enrollment:
return CourseEnrollmentResponseStatuses.CONFLICT
enrollment_status = ProgramCourseEnrollment.enroll(
return ProgramCourseEnrollment.enroll(
program_enrollment,
self.course_key,
enrollment_request['status']
)
return enrollment_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'])

34
lms/djangoapps/program_enrollments/models.py
View File

@@ -3,6 +3,7 @@
Django model specifications for the Program Enrollments API
"""
from __future__ import unicode_literals
import logging
from django.contrib.auth.models import User
from django.core.exceptions import ValidationError
@@ -15,6 +16,8 @@ from opaque_keys.edx.django.models import CourseKeyField
from simple_history.models import HistoricalRecords
from student.models import CourseEnrollment as StudentCourseEnrollment
logger = logging.getLogger(__name__) # pylint: disable=invalid-name
class ProgramEnrollment(TimeStampedModel): # pylint: disable=model-missing-unicode
"""
@@ -162,3 +165,34 @@ class ProgramCourseEnrollment(TimeStampedModel): # pylint: disable=model-missin
status=status,
)
return program_course_enrollment.status
def change_status(self, status):
"""
Modify ProgramCourseEnrollment status and course_enrollment status if it exists
"""
if status == self.status:
return status
self.status = status
if self.course_enrollment:
if status == CourseEnrollmentResponseStatuses.ACTIVE:
self.course_enrollment.activate()
elif status == CourseEnrollmentResponseStatuses.INACTIVE:
self.course_enrollment.deactivate()
else:
message = ("Changed {enrollment} status to {status}, not changing course_enrollment"
" status because status is not '{active}' or '{inactive}'")
logger.warn(message.format(
enrollment=self,
status=status,
active=CourseEnrollmentResponseStatuses.ACTIVE,
inactive=CourseEnrollmentResponseStatuses.INACTIVE
))
elif self.program_enrollment.user:
logger.warn("User {user} {program_enrollment} {course_key} has no course_enrollment".format(
user=self.program_enrollment.user,
program_enrollment=self.program_enrollment,
course_key=self.course_key,
))
self.save()
return self.status

62
lms/djangoapps/program_enrollments/tests/test_models.py
View File

@@ -4,11 +4,14 @@ Unit tests for ProgramEnrollment models.
from __future__ import unicode_literals
from uuid import uuid4
from testfixtures import LogCapture
from django.test import TestCase
from opaque_keys.edx.keys import CourseKey
from lms.djangoapps.program_enrollments.models import ProgramEnrollment
from student.tests.factories import UserFactory
from lms.djangoapps.program_enrollments.models import ProgramEnrollment, ProgramCourseEnrollment
from student.tests.factories import UserFactory, CourseEnrollmentFactory
from openedx.core.djangoapps.catalog.tests.factories import generate_course_run_key
class ProgramEnrollmentModelTests(TestCase):
@@ -97,3 +100,58 @@ class ProgramEnrollmentModelTests(TestCase):
self.assertTrue(self.enrollment.historical_records.all())
for record in self.enrollment.historical_records.all():
self.assertEquals(record.external_user_key, None)
class ProgramCourseEnrollmentModelTests(TestCase):
"""
Tests for the ProgramCourseEnrollment model.
"""
def setUp(self):
"""
Set up test data
"""
super(ProgramCourseEnrollmentModelTests, self).setUp()
self.user = UserFactory.create()
self.program_uuid = uuid4()
self.program_enrollment = ProgramEnrollment.objects.create(
user=self.user,
external_user_key='abc',
program_uuid=self.program_uuid,
curriculum_uuid=uuid4(),
status='enrolled'
)
self.course_key = CourseKey.from_string(generate_course_run_key())
self.course_enrollment = CourseEnrollmentFactory.create(
course_id=self.course_key,
user=self.user,
)
self.program_course_enrollment = ProgramCourseEnrollment.objects.create(
program_enrollment=self.program_enrollment,
course_key=self.course_key,
course_enrollment=self.course_enrollment,
status="active"
)
def test_change_status_no_enrollment(self):
with LogCapture() as capture:
self.program_course_enrollment.course_enrollment = None
self.program_course_enrollment.change_status("inactive")
expected_message = "User {} {} {} has no course_enrollment".format(
self.user,
self.program_enrollment,
self.course_key
)
capture.check(
('lms.djangoapps.program_enrollments.models', 'WARNING', expected_message)
)
def test_change_status_not_active_or_inactive(self):
with LogCapture() as capture:
status = "potential-future-status-0123"
self.program_course_enrollment.change_status(status)
message = ("Changed {} status to {}, not changing course_enrollment"
" status because status is not 'active' or 'inactive'")
expected_message = message.format(self.program_course_enrollment, status)
capture.check(
('lms.djangoapps.program_enrollments.models', 'WARNING', expected_message)
)