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

Write course_staff role on enrollment (#23677)

Browse Source
This commit is contained in:
Zachary Hancock
2020-04-13 14:13:24 -04:00
committed by GitHub
parent add68c440a
commit f0d7ff9025
4 changed files with 604 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files

438
lms/djangoapps/program_enrollments/api/tests/test_writing.py
View File

@@ -11,48 +11,117 @@ mocks in the view tests.
from uuid import UUID
import ddt
from django.core.cache import cache
from opaque_keys.edx.keys import CourseKey
from organizations.tests.factories import OrganizationFactory
from course_modes.models import CourseMode
from lms.djangoapps.program_enrollments.constants import ProgramCourseEnrollmentRoles
from lms.djangoapps.program_enrollments.constants import ProgramCourseOperationStatuses as CourseStatuses
from lms.djangoapps.program_enrollments.constants import ProgramEnrollmentStatuses as PEStatuses
from lms.djangoapps.program_enrollments.models import ProgramEnrollment
from lms.djangoapps.program_enrollments.models import (
CourseAccessRoleAssignment,
ProgramCourseEnrollment,
ProgramEnrollment
)
from lms.djangoapps.program_enrollments.tests.factories import ProgramCourseEnrollmentFactory, ProgramEnrollmentFactory
from openedx.core.djangoapps.catalog.cache import PROGRAM_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.tests.factories import CourseOverviewFactory
from openedx.core.djangolib.testing.utils import CacheIsolationTestCase
from student.roles import CourseStaffRole
from student.tests.factories import CourseEnrollmentFactory, UserFactory
from third_party_auth.tests.factories import SAMLProviderConfigFactory
from ..writing import write_program_enrollments
from ..writing import write_program_course_enrollments, write_program_enrollments
class WritingProgramEnrollmentTest(CacheIsolationTestCase):
class EnrollmentTestMixin(CacheIsolationTestCase):
"""
Test cases for program enrollment writing functions.
Test data and helper functions
"""
ENABLED_CACHES = ['default']
organization_key = 'test'
program_uuid_x = UUID('dddddddd-5f48-493d-9910-84e1d36c657f')
program_uuid = UUID('dddddddd-5f48-493d-9910-84e1d36c657f')
curriculum_uuid_a = UUID('aaaaaaaa-bd26-4370-94b8-b4063858210b')
user_0 = 'user-0'
def setUp(self):
@classmethod
def setUpClass(cls):
"""
Set up test data
"""
super(WritingProgramEnrollmentTest, self).setUp()
catalog_org = CatalogOrganizationFactory.create(key=self.organization_key)
program = ProgramFactory.create(
uuid=self.program_uuid_x,
super(EnrollmentTestMixin, cls).setUpClass()
catalog_org = CatalogOrganizationFactory.create(key=cls.organization_key)
cls.program = ProgramFactory.create(
uuid=cls.program_uuid,
authoring_organizations=[catalog_org]
)
organization = OrganizationFactory.create(short_name=self.organization_key)
organization = OrganizationFactory.create(short_name=cls.organization_key)
SAMLProviderConfigFactory.create(organization=organization)
cache.set(PROGRAM_CACHE_KEY_TPL.format(uuid=self.program_uuid_x), program, None)
catalog_course_id_str = 'course-v1:edX+ToyX'
course_run_id_str = '{}+Toy_Course'.format(catalog_course_id_str)
cls.course_id = CourseKey.from_string(course_run_id_str)
CourseOverviewFactory(id=cls.course_id)
course_run = CourseRunFactory(key=course_run_id_str)
cls.course = CourseFactory(key=catalog_course_id_str, course_runs=[course_run])
cls.student_1 = UserFactory(username='student-1')
cls.student_2 = UserFactory(username='student-2')
def setUp(self):
super(EnrollmentTestMixin, self).setUp()
cache.set(PROGRAM_CACHE_KEY_TPL.format(uuid=self.program_uuid), self.program, None)
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=CourseStatuses.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_id,
user=program_enrollment.user,
mode=CourseMode.MASTERS
)
course_enrollment.is_active = course_status == CourseStatuses.ACTIVE
course_enrollment.save()
return ProgramCourseEnrollmentFactory.create(
program_enrollment=program_enrollment,
course_key=self.course_id,
course_enrollment=course_enrollment,
status=course_status,
)
def create_program_and_course_enrollments(self, external_user_key, user=False, course_status=CourseStatuses.ACTIVE):
program_enrollment = self.create_program_enrollment(external_user_key, user)
return self.create_program_course_enrollment(program_enrollment, course_status=course_status)
class WritingProgramEnrollmentTest(EnrollmentTestMixin):
"""
Test cases for program enrollment writing functions.
"""
def test_write_program_enrollments_status_ended(self):
"""
Successfully updates program enrollment to status ended if requested.
@@ -60,18 +129,351 @@ class WritingProgramEnrollmentTest(CacheIsolationTestCase):
"""
assert ProgramEnrollment.objects.count() == 0
assert ProgramEnrollment.historical_records.count() == 0 # pylint: disable=no-member
write_program_enrollments(self.program_uuid_x, [{
'external_user_key': self.user_0,
write_program_enrollments(self.program_uuid, [{
'external_user_key': 'learner-1',
'status': PEStatuses.PENDING,
'curriculum_uuid': self.curriculum_uuid_a,
}], True, False)
assert ProgramEnrollment.objects.count() == 1
assert ProgramEnrollment.historical_records.count() == 1 # pylint: disable=no-member
write_program_enrollments(self.program_uuid_x, [{
'external_user_key': self.user_0,
write_program_enrollments(self.program_uuid, [{
'external_user_key': 'learner-1',
'status': PEStatuses.ENDED,
'curriculum_uuid': self.curriculum_uuid_a,
}], False, True)
assert ProgramEnrollment.objects.count() == 1
assert ProgramEnrollment.historical_records.count() == 2 # pylint: disable=no-member
assert ProgramEnrollment.objects.filter(status=PEStatuses.ENDED).exists()
@ddt.ddt
class WriteProgramCourseEnrollmentTest(EnrollmentTestMixin):
""" Test write_program_enrollments API """
def course_enrollment_request(self, external_key, status=CourseStatuses.ACTIVE, course_staff=None):
"""
Constructs a single course enrollment request object
"""
return {
'external_user_key': external_key,
'status': status,
'course_staff': course_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_id, 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_id, course_enrollment.course_id)
self.assertEqual(mode, course_enrollment.mode)
else:
self.assertIsNone(course_enrollment)
def setup_change_test_data(self, initial_statuses):
"""
Helper method to setup initial state for update tests
"""
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)
def test_create_only(self):
"""
Test creating new program course enrollments with only the create flag true
"""
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)
course_enrollment_requests = [
self.course_enrollment_request('learner-1', CourseStatuses.ACTIVE),
self.course_enrollment_request('learner-2', CourseStatuses.INACTIVE),
self.course_enrollment_request('learner-3', CourseStatuses.ACTIVE),
self.course_enrollment_request('learner-4', CourseStatuses.INACTIVE),
]
result = write_program_course_enrollments(
self.program_uuid,
self.course_id,
course_enrollment_requests,
True,
False
)
self.assertDictEqual(
{
'learner-1': CourseStatuses.ACTIVE,
'learner-2': CourseStatuses.INACTIVE,
'learner-3': CourseStatuses.ACTIVE,
'learner-4': CourseStatuses.INACTIVE,
},
result,
)
self.assert_program_course_enrollment('learner-1', CourseStatuses.ACTIVE, True)
self.assert_program_course_enrollment('learner-2', CourseStatuses.INACTIVE, True)
self.assert_program_course_enrollment('learner-3', CourseStatuses.ACTIVE, False)
self.assert_program_course_enrollment('learner-4', CourseStatuses.INACTIVE, False)
@ddt.data(
('active', 'inactive', 'active', 'inactive'),
('inactive', 'active', 'inactive', 'active'),
('active', 'active', 'active', 'active'),
('inactive', 'inactive', 'inactive', 'inactive'),
)
def test_update_only(self, initial_statuses):
"""
Test updating existing enrollments with only the update flag true
"""
self.setup_change_test_data(initial_statuses)
course_enrollment_requests = [
self.course_enrollment_request('learner-1', CourseStatuses.INACTIVE),
self.course_enrollment_request('learner-2', CourseStatuses.ACTIVE),
self.course_enrollment_request('learner-3', CourseStatuses.INACTIVE),
self.course_enrollment_request('learner-4', CourseStatuses.ACTIVE),
]
result = write_program_course_enrollments(
self.program_uuid,
self.course_id,
course_enrollment_requests,
False,
True,
)
self.assertDictEqual(
{
'learner-1': CourseStatuses.INACTIVE,
'learner-2': CourseStatuses.ACTIVE,
'learner-3': CourseStatuses.INACTIVE,
'learner-4': CourseStatuses.ACTIVE,
},
result,
)
self.assert_program_course_enrollment('learner-1', CourseStatuses.INACTIVE, True)
self.assert_program_course_enrollment('learner-2', CourseStatuses.ACTIVE, True)
self.assert_program_course_enrollment('learner-3', CourseStatuses.INACTIVE, False)
self.assert_program_course_enrollment('learner-4', CourseStatuses.ACTIVE, False)
def test_create_or_update(self):
"""
Test writing enrollments with both create and update flags true.
Existing enrollments should be updated. If no matching enrollment is found, create one.
"""
# learners 1-4 are already enrolled in courses, 5-6 only have a program enrollment
self.setup_change_test_data([
CourseStatuses.ACTIVE, CourseStatuses.ACTIVE,
CourseStatuses.ACTIVE, CourseStatuses.ACTIVE]
)
self.create_program_enrollment('learner-5')
self.create_program_enrollment('learner-6', user=None)
course_enrollment_requests = [
self.course_enrollment_request('learner-1', CourseStatuses.INACTIVE),
self.course_enrollment_request('learner-2', CourseStatuses.ACTIVE),
self.course_enrollment_request('learner-5', CourseStatuses.ACTIVE),
self.course_enrollment_request('learner-6', CourseStatuses.ACTIVE),
]
result = write_program_course_enrollments(
self.program_uuid,
self.course_id,
course_enrollment_requests,
True,
True,
)
self.assertDictEqual(
{
'learner-1': CourseStatuses.INACTIVE,
'learner-2': CourseStatuses.ACTIVE,
'learner-5': CourseStatuses.ACTIVE,
'learner-6': CourseStatuses.ACTIVE,
},
result,
)
self.assert_program_course_enrollment('learner-1', CourseStatuses.INACTIVE, True)
self.assert_program_course_enrollment('learner-2', CourseStatuses.ACTIVE, True)
self.assert_program_course_enrollment('learner-5', CourseStatuses.ACTIVE, True)
self.assert_program_course_enrollment('learner-6', CourseStatuses.ACTIVE, False)
def test_create_conflicting_enrollment(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')
course_enrollment_requests = [self.course_enrollment_request('learner-1')]
result = write_program_course_enrollments(
self.program_uuid,
self.course_id,
course_enrollment_requests,
True,
False,
)
self.assertDictEqual({'learner-1': CourseStatuses.CONFLICT}, result)
def test_update_nonexistent_enrollment(self):
self.create_program_enrollment('learner-1')
result = write_program_course_enrollments(
self.program_uuid,
self.course_id,
[self.course_enrollment_request('learner-1')],
False,
True,
)
self.assertDictEqual({'learner-1': CourseStatuses.NOT_FOUND}, result)
def test_invalid_status(self):
self.create_program_enrollment('learner-1')
result = write_program_course_enrollments(
self.program_uuid,
self.course_id,
[self.course_enrollment_request('learner-1', 'not-a-status')],
True,
False,
)
self.assertDictEqual({'learner-1': CourseStatuses.INVALID_STATUS}, result)
def test_duplicate_external_keys(self):
self.create_program_enrollment('learner-1')
course_enrollment_requests = [
self.course_enrollment_request('learner-1'),
self.course_enrollment_request('learner-1'),
]
result = write_program_course_enrollments(
self.program_uuid,
self.course_id,
course_enrollment_requests,
True,
False,
)
self.assertDictEqual(
{
'learner-1': CourseStatuses.DUPLICATED,
},
result
)
def test_learner_not_in_program(self):
result = write_program_course_enrollments(
self.program_uuid,
self.course_id,
[self.course_enrollment_request('learner-1', CourseStatuses.ACTIVE)],
True,
False,
)
self.assertDictEqual({'learner-1': CourseStatuses.NOT_IN_PROGRAM}, result)
def test_create_enrollments_and_assign_staff(self):
"""
Successfully creates both waiting and linked program course enrollments with the course staff role.
"""
course_staff_role = CourseStaffRole(self.course_id)
course_staff_role.add_users(self.student_1)
self.create_program_enrollment('learner-1', user=None)
self.create_program_enrollment('learner-2', user=self.student_1)
self.create_program_enrollment('learner-3', user=self.student_2)
course_enrollment_requests = [
self.course_enrollment_request('learner-1', CourseStatuses.ACTIVE, True),
self.course_enrollment_request('learner-2', CourseStatuses.ACTIVE, True),
self.course_enrollment_request('learner-3', CourseStatuses.ACTIVE, True),
]
write_program_course_enrollments(
self.program_uuid,
self.course_id,
course_enrollment_requests,
True,
True,
)
self.assert_program_course_enrollment('learner-1', CourseStatuses.ACTIVE, False)
self.assert_program_course_enrollment('learner-2', CourseStatuses.ACTIVE, True)
self.assert_program_course_enrollment('learner-3', CourseStatuses.ACTIVE, True)
# Users linked to either enrollment are given the course staff role
self.assertListEqual(
[self.student_1, self.student_2],
list(course_staff_role.users_with_role())
)
# CourseAccessRoleAssignment objects are created for enrollments with no linked user
pending_role_assingments = CourseAccessRoleAssignment.objects.all()
assert pending_role_assingments.count() == 1
pending_role_assingments.get(
enrollment__program_enrollment__external_user_key='learner-1',
enrollment__course_key=self.course_id
)
def test_update_and_assign_or_revoke_staff(self):
"""
Successfully updates existing enrollments to assign or revoke the CourseStaff role.
"""
course_staff_role = CourseStaffRole(self.course_id)
course_staff_role.add_users(self.student_2)
self.create_program_and_course_enrollments('learner-1', user=self.student_1)
self.create_program_and_course_enrollments('learner-2', user=self.student_2)
self.create_program_and_course_enrollments('learner-3', user=None)
enrollment = self.create_program_and_course_enrollments('learner-4', user=None)
CourseAccessRoleAssignment.objects.create(
enrollment=enrollment,
role=ProgramCourseEnrollmentRoles.COURSE_STAFF,
)
course_enrollment_requests = [
self.course_enrollment_request('learner-1', CourseStatuses.ACTIVE, True),
self.course_enrollment_request('learner-2', CourseStatuses.ACTIVE, False),
self.course_enrollment_request('learner-3', CourseStatuses.ACTIVE, True),
self.course_enrollment_request('learner-4', CourseStatuses.ACTIVE, False),
]
write_program_course_enrollments(
self.program_uuid,
self.course_id,
course_enrollment_requests,
True,
True,
)
# Role is revoked for user's with a linked enrollment
self.assertListEqual(
[self.student_1],
list(course_staff_role.users_with_role())
)
# CourseAccessRoleAssignment objects are created/revoked for enrollments with no linked user
pending_role_assingments = CourseAccessRoleAssignment.objects.all()
assert pending_role_assingments.count() == 1
pending_role_assingments.get(
enrollment__program_enrollment__external_user_key='learner-3',
enrollment__course_key=self.course_id
)
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_id,
user=self.student_1,
mode=CourseMode.VERIFIED
)
self.create_program_enrollment('learner-1', user=self.student_1)
write_program_course_enrollments(
self.program_uuid,
self.course_id,
[self.course_enrollment_request('learner-1', CourseStatuses.ACTIVE)],
True,
False
)
self.assert_program_course_enrollment('learner-1', CourseStatuses.ACTIVE, True, mode=CourseMode.VERIFIED)

77
lms/djangoapps/program_enrollments/api/writing.py
View File

@@ -13,13 +13,14 @@ from simple_history.utils import bulk_create_with_history
from course_modes.models import CourseMode
from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
from student.models import CourseEnrollment, NonExistentCourseError
from student.roles import CourseStaffRole
from ..constants import ProgramCourseEnrollmentStatuses
from ..constants import ProgramCourseEnrollmentRoles, ProgramCourseEnrollmentStatuses
from ..constants import ProgramCourseOperationStatuses as ProgramCourseOpStatuses
from ..constants import ProgramEnrollmentStatuses
from ..constants import ProgramOperationStatuses as ProgramOpStatuses
from ..exceptions import ProviderDoesNotExistException
from ..models import ProgramCourseEnrollment, ProgramEnrollment
from ..models import CourseAccessRoleAssignment, ProgramCourseEnrollment, ProgramEnrollment
from .reading import fetch_program_course_enrollments_by_students, fetch_program_enrollments, get_users_by_external_keys
logger = logging.getLogger(__name__)
@@ -181,6 +182,7 @@ def write_program_course_enrollments(
enrollment_requests (list[dict]): dicts in the form:
* 'external_user_key': str
* 'status': str from ProgramCourseEnrollmentStatuses
* 'course_staff': Boolean if the user should have the CourseStaff role
create (bool): non-existent enrollments will be created iff `create`,
otherwise they will be skipped as 'duplicate'.
update (bool): existing enrollments will be updated iff `update`,
@@ -245,10 +247,14 @@ def write_program_course_enrollments(
# For each enrollment request, try to create/update.
# For creates, build up list `to_save`, which we will bulk-create afterwards.
# For updates, do them in place (Django 2.2 will add bulk-update support).
# For updates, do them in place in order to preserve history records.
# For each operation, update `results` with the new status or an error status.
to_save = []
enrollments_to_save = []
created_enrollments = []
updated_enrollments = []
staff_assignments_by_user_key = {}
for external_key, request in requests_by_key.items():
course_staff = request['course_staff']
status = request['status']
if status not in ProgramCourseEnrollmentStatuses.__ALL__:
results[external_key] = ProgramCourseOpStatuses.INVALID_STATUS
@@ -265,6 +271,7 @@ def write_program_course_enrollments(
results[external_key] = change_program_course_enrollment_status(
existing_course_enrollment, status
)
updated_enrollments.append(existing_course_enrollment)
else:
if not create:
results[external_key] = ProgramCourseOpStatuses.NOT_FOUND
@@ -272,14 +279,27 @@ def write_program_course_enrollments(
new_course_enrollment = create_program_course_enrollment(
program_enrollment, course_key, status, save=False
)
to_save.append(new_course_enrollment)
enrollments_to_save.append(new_course_enrollment)
results[external_key] = new_course_enrollment.status
if course_staff is not None:
staff_assignments_by_user_key[external_key] = course_staff
# Bulk-create all new program-course enrollments and corresponding history records.
# Note: this will NOT invoke `save()` or `pre_save`/`post_save` signals!
# See https://docs.djangoproject.com/en/1.11/ref/models/querysets/#bulk-create.
if to_save:
bulk_create_with_history(to_save, ProgramCourseEnrollment)
if enrollments_to_save:
created_enrollments = bulk_create_with_history(enrollments_to_save, ProgramCourseEnrollment)
# For every created/updated enrollment, check if the user should be course staff.
# If that enrollment has a linked user, assign the user the course staff role
# If that enrollment does not have a linked user, create a CourseAccessRoleAssignment
# for that enrollment.
written_enrollments = ProgramCourseEnrollment.objects.filter(
id__in=[enrollment.id for enrollment in created_enrollments + updated_enrollments]
).select_related('program_enrollment')
_assign_course_staff_role(course_key, written_enrollments, staff_assignments_by_user_key)
return results
@@ -407,6 +427,49 @@ def enroll_in_masters_track(user, course_key, status):
return course_enrollment
def _assign_course_staff_role(course_key, enrollments, staff_assignments):
"""
Grant or remove the course staff role for a set of enrollments on a course.
For enrollment without a linked user, a CourseAccessRoleAssignment will be
created (or removed) for that enrollment.
Arguments:
enrollments (list): ProgramCourseEnrollments to update
staff_assignments (dict): Maps an enrollment's external key to a course staff value
"""
role_assignments_to_save = []
enrollment_role_assignments_to_delete = []
for enrollment in enrollments:
if enrollment.course_key != course_key:
continue
external_key = enrollment.program_enrollment.external_user_key
user = enrollment.program_enrollment.user
course_staff = staff_assignments.get(external_key)
if user:
if course_staff is True:
CourseStaffRole(course_key).add_users(user)
elif course_staff is False:
CourseStaffRole(course_key).remove_users(user)
else:
if course_staff is True:
role_assignments_to_save.append(CourseAccessRoleAssignment(
enrollment=enrollment,
role=ProgramCourseEnrollmentRoles.COURSE_STAFF
))
elif course_staff is False:
enrollment_role_assignments_to_delete.append(enrollment)
if role_assignments_to_save:
CourseAccessRoleAssignment.objects.bulk_create(role_assignments_to_save)
if enrollment_role_assignments_to_delete:
CourseAccessRoleAssignment.objects.filter(
enrollment__in=enrollment_role_assignments_to_delete
).delete()
def _ensure_course_exists(course_key, user_key_or_id):
"""
Log and raise an error if `course_key` does not refer to a real course run.

1
lms/djangoapps/program_enrollments/rest_api/v1/serializers.py
View File

@@ -102,6 +102,7 @@ class ProgramCourseEnrollmentRequestSerializer(serializers.Serializer, InvalidSt
# returning INVALID_STATUS for individual bad statuses instead of raising
# a ValidationError for the entire request.
status = serializers.CharField(allow_blank=False)
course_staff = serializers.BooleanField(required=False, default=None)
class ProgramCourseGradeSerializer(serializers.Serializer):

335
lms/djangoapps/program_enrollments/rest_api/v1/tests/test_views.py
View File

@@ -4,7 +4,7 @@ Unit tests for ProgramEnrollment views.
import json
from collections import defaultdict
from collections import defaultdict, OrderedDict
from datetime import datetime, timedelta
from uuid import UUID, uuid4
@@ -153,17 +153,14 @@ class EnrollmentsDataMixin(ProgramCacheMixin):
def log_in_staff(self):
self.client.login(username=self.global_staff.username, password=self.password)
def learner_enrollment(self, student_key, enrollment_status="active"):
def learner_enrollment(self, student_key, enrollment_status="active", course_staff=None):
"""
Convenience method to create a learner enrollment record
"""
return {"student_key": student_key, "status": enrollment_status}
def request(self, path, data, **kwargs):
pass
def prepare_student(self, key):
pass
enrollment_record = {"student_key": student_key, "status": enrollment_status}
if course_staff is not None:
enrollment_record["course_staff"] = course_staff
return enrollment_record
def create_program_enrollment(self, external_user_key, user=False):
"""
@@ -766,6 +763,7 @@ class ProgramCourseEnrollmentsMixin(EnrollmentsDataMixin):
Children should override self.request()
"""
view_name = 'programs_api:v1:program_course_enrollments'
write_enrollments_function = ''
@classmethod
def setUpClass(cls):
@@ -782,26 +780,6 @@ class ProgramCourseEnrollmentsMixin(EnrollmentsDataMixin):
self.default_url = self.get_url(course_id=self.course_id)
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_id, 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_id, 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")]
@@ -839,40 +817,6 @@ class ProgramCourseEnrollmentsMixin(EnrollmentsDataMixin):
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):
self.prepare_student('learner-1')
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'}],
@@ -896,17 +840,42 @@ class ProgramCourseEnrollmentsMixin(EnrollmentsDataMixin):
self.assertEqual(response.status_code, 400)
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)
mock_write_response = {'learner-1': 'inactive'}
with mock.patch(
_VIEW_PATCH_FORMAT.format('write_program_course_enrollments'),
autospec=True,
return_value=mock_write_response,
):
response = self.request(self.default_url, request_data)
self.assertEqual(200, response.status_code)
self.assertDictEqual(
mock_write_response,
response.data,
)
self.assertEqual(response.status_code, 200)
self.assertDictEqual(
response.data,
{'learner-1': 'inactive'}
)
def test_207_multistatus(self):
"""
If errors occur but some requests succeed return a 207
"""
request_data = [self.learner_enrollment("learner-1"), self.learner_enrollment("learner-2")]
mock_write_response = {
'learner-1': CourseStatuses.ACTIVE,
'learner-2': CourseStatuses.NOT_IN_PROGRAM,
}
with mock.patch(
_VIEW_PATCH_FORMAT.format('write_program_course_enrollments'),
autospec=True,
return_value=mock_write_response,
):
response = self.request(self.default_url, request_data)
self.assertEqual(207, response.status_code)
self.assertDictEqual(
{'learner-1': CourseStatuses.ACTIVE, 'learner-2': CourseStatuses.NOT_IN_PROGRAM},
response.data
)
class ProgramCourseEnrollmentsGetTests(EnrollmentsDataMixin, APITestCase):
@@ -1043,164 +1012,103 @@ class ProgramCourseEnrollmentsPostTests(ProgramCourseEnrollmentsMixin, APITestCa
def request(self, path, data, **kwargs):
return self.client.post(path, data, format='json', **kwargs)
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)
mock_write_response = {
"learner-1": "active",
"learner-2": "inactive",
"learner-3": "active",
"learner-4": "inactive",
}
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"),
self.learner_enrollment("learner-3", "active", True),
self.learner_enrollment("learner-4", "inactive", False),
]
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
"""
with mock.patch(
_VIEW_PATCH_FORMAT.format('write_program_course_enrollments'),
autospec=True,
return_value=mock_write_response,
) as mock_write:
response = self.request(self.default_url, post_data)
self.assertEqual(200, response.status_code)
self.assertDictEqual(
mock_write_response,
response.data,
)
mock_write.assert_called_once_with(
str(self.program_uuid),
self.course_id,
[
OrderedDict([('external_user_key', 'learner-1'), ('status', 'active'), ('course_staff', None)]),
OrderedDict([('external_user_key', 'learner-2'), ('status', 'inactive'), ('course_staff', None)]),
OrderedDict([('external_user_key', 'learner-3'), ('status', 'active'), ('course_staff', True)]),
OrderedDict([('external_user_key', 'learner-4'), ('status', 'inactive'), ('course_staff', False)]),
],
create=True,
update=False,
)
def test_no_successful_enrollments(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_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_id,
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
)
mock_write_response = {'learner-1': CourseStatuses.CONFLICT}
with mock.patch(
_VIEW_PATCH_FORMAT.format('write_program_course_enrollments'),
autospec=True,
return_value=mock_write_response,
):
response = self.request(self.default_url, post_data)
self.assertEqual(422, response.status_code)
self.assertDictEqual({'learner-1': CourseStatuses.CONFLICT}, response.data)
@ddt.ddt
class ProgramCourseEnrollmentsModifyMixin(ProgramCourseEnrollmentsMixin):
"""
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_update_enrollment(self):
request_data = [self.learner_enrollment('learner-1', 'active')]
mock_write_response = {'learner-1': 'active'}
with mock.patch(
_VIEW_PATCH_FORMAT.format('write_program_course_enrollments'),
autospec=True,
return_value=mock_write_response,
) as mock_write:
response = self.request(self.default_url, request_data)
self.assertEqual(200, response.status_code)
self.assertDictEqual(
mock_write_response,
response.data,
)
mock_write.assert_called_once_with(
str(self.program_uuid),
self.course_id,
[
OrderedDict([('external_user_key', 'learner-1'), ('status', 'active'), ('course_staff', None)])
],
create=self.create,
update=self.update,
)
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 ProgramCourseEnrollmentsPatchTests(ProgramCourseEnrollmentsModifyMixin, APITestCase):
""" Tests for course enrollment PATCH """
""" Test PATCH ProgramCourseEnrollments """
update = True
create = False
def request(self, path, data, **kwargs):
return self.client.patch(path, data, format='json', **kwargs)
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 ProgramCourseEnrollmentsPutTests(ProgramCourseEnrollmentsModifyMixin, APITestCase):
""" Test PUT ProgramCourseEnrollments """
update = True
create = True
def request(self, path, data, **kwargs):
return self.client.put(path, data, format='json', **kwargs)
@ddt.ddt
@@ -1369,23 +1277,6 @@ class MultiprogramEnrollmentsTest(EnrollmentsDataMixin, APITestCase):
self.assertDictEqual(expected_results, response.data)
class ProgramCourseEnrollmentsPutTests(ProgramCourseEnrollmentsModifyMixin, APITestCase):
""" Tests for course enrollment PUT """
def request(self, path, data, **kwargs):
return self.client.put(path, data, format='json', **kwargs)
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 ProgramCourseGradesGetTests(EnrollmentsDataMixin, APITestCase):
"""
Tests for GET calls to the Program Course Grades API.