Merge pull request #23720 from edx/mroytman/MST-184-enrollments-linking-course-staff
Realize CourseAccessRoles for a user when an LMS user is linked to th…
This commit is contained in:
@@ -1,7 +1,14 @@
|
||||
# pylint: disable=unused-import
|
||||
"""
|
||||
Python APIs exposed by the student app to other in-process apps.
|
||||
"""
|
||||
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.conf import settings
|
||||
|
||||
from student.models_api import create_manual_enrollment_audit as _create_manual_enrollment_audit
|
||||
from student.models_api import get_course_access_role
|
||||
from student.models_api import get_course_enrollment as _get_course_enrollment
|
||||
from student.models_api import (
|
||||
ENROLLED_TO_ENROLLED as _ENROLLED_TO_ENROLLED,
|
||||
@@ -13,6 +20,7 @@ from student.models_api import (
|
||||
ALLOWEDTOENROLL_TO_UNENROLLED as _ALLOWEDTOENROLL_TO_UNENROLLED,
|
||||
DEFAULT_TRANSITION_STATE as _DEFAULT_TRANSITION_STATE,
|
||||
)
|
||||
from student.roles import REGISTERED_ACCESS_ROLES as _REGISTERED_ACCESS_ROLES
|
||||
from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers
|
||||
|
||||
|
||||
@@ -90,3 +98,16 @@ def create_manual_enrollment_audit(
|
||||
enrollment,
|
||||
role
|
||||
)
|
||||
|
||||
|
||||
def get_access_role_by_role_name(role_name):
|
||||
"""
|
||||
Get the concrete child class of the AccessRole abstract class associated with the string role_name
|
||||
by looking in REGISTERED_ACCESS_ROLES. If there is no class associated with this name, return None.
|
||||
|
||||
Note that this will only return classes that are registered in _REGISTERED_ACCESS_ROLES.
|
||||
|
||||
Arguments:
|
||||
role_name: the name of the role
|
||||
"""
|
||||
return _REGISTERED_ACCESS_ROLES.get(role_name, None)
|
||||
|
||||
@@ -3,6 +3,7 @@ Provides Python APIs exposed from Student models.
|
||||
"""
|
||||
import logging
|
||||
|
||||
from student.models import CourseAccessRole as _CourseAccessRole
|
||||
from student.models import CourseEnrollment as _CourseEnrollment
|
||||
from student.models import ManualEnrollmentAudit as _ManualEnrollmentAudit
|
||||
from student.models import (
|
||||
@@ -62,3 +63,33 @@ def get_phone_number(user_id):
|
||||
log.exception(exception)
|
||||
return None
|
||||
return student.phone_number or None
|
||||
|
||||
|
||||
def get_course_access_role(user, org, course_id, role):
|
||||
"""
|
||||
Get a specific CourseAccessRole object. Return None if
|
||||
it does not exist.
|
||||
|
||||
Arguments:
|
||||
user: User object for the user who has access in a course
|
||||
org: the org the course is in
|
||||
course_id: the course_id of the CourseAccessRole
|
||||
role: the role type of the role
|
||||
"""
|
||||
try:
|
||||
course_access_role = _CourseAccessRole.objects.get(
|
||||
user=user,
|
||||
org=org,
|
||||
course_id=course_id,
|
||||
role=role,
|
||||
)
|
||||
except _CourseAccessRole.DoesNotExist:
|
||||
log.exception('No CourseAccessRole found for user_id=%(user_id)s, org=%(org)s, '
|
||||
'course_id=%(course_id)s, and role=%(role)s.', {
|
||||
'user': user.id,
|
||||
'org': org,
|
||||
'course_id': course_id,
|
||||
'role': role,
|
||||
})
|
||||
return None
|
||||
return course_access_role
|
||||
|
||||
@@ -12,6 +12,7 @@ import logging
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.db import IntegrityError, transaction
|
||||
|
||||
from student.api import get_access_role_by_role_name
|
||||
from student.models import CourseEnrollmentException
|
||||
|
||||
from .reading import fetch_program_enrollments
|
||||
@@ -159,7 +160,7 @@ def link_program_enrollment_to_lms_user(program_enrollment, user):
|
||||
program_enrollment.external_user_key,
|
||||
program_enrollment.program_uuid,
|
||||
)
|
||||
logger.info("Linking " + link_log_info)
|
||||
logger.info("Linking %s", link_log_info)
|
||||
program_enrollment.user = user
|
||||
try:
|
||||
program_enrollment.save()
|
||||
@@ -169,8 +170,9 @@ def link_program_enrollment_to_lms_user(program_enrollment, user):
|
||||
user, pce.course_key, pce.status
|
||||
)
|
||||
pce.save()
|
||||
_fulfill_course_access_roles(user, pce)
|
||||
except IntegrityError:
|
||||
logger.error("Integrity error while linking " + link_log_info)
|
||||
logger.error("Integrity error while linking %s", link_log_info)
|
||||
raise
|
||||
except CourseEnrollmentException as e:
|
||||
logger.error(
|
||||
@@ -179,3 +181,53 @@ def link_program_enrollment_to_lms_user(program_enrollment, user):
|
||||
)
|
||||
)
|
||||
raise
|
||||
|
||||
|
||||
def _fulfill_course_access_roles(user, program_course_enrollment):
|
||||
"""
|
||||
Convert any CourseAccessRoleAssignment objects, which represent pending CourseAccessRoles, into fulfilled
|
||||
CourseAccessRole objects as part of a program course enrollment.
|
||||
|
||||
Arguments:
|
||||
user: User object for whom we are fulfilling CourseAccessRoleAssignments into CourseAccessRoles
|
||||
program_course_enrollment: the ProgramCourseEnrollment object that represents the course the user
|
||||
should be granted a CourseAccessRole in the context of
|
||||
"""
|
||||
role_assignments = program_course_enrollment.courseaccessroleassignment_set.all()
|
||||
program_enrollment = program_course_enrollment.program_enrollment
|
||||
|
||||
for role_assignment in role_assignments:
|
||||
# currently, we only allow for an assignment of a "staff" role, but
|
||||
# this allows us to expand the functionality to other roles, if need be
|
||||
# get_access_role_by_role_name gets us the class, so we need to instantiate it
|
||||
role = get_access_role_by_role_name(role_assignment.role)(program_course_enrollment.course_key)
|
||||
|
||||
logger_format_values = {
|
||||
'course_key': program_course_enrollment.course_key,
|
||||
'program_course_enrollment': program_course_enrollment,
|
||||
'program_uuid': program_enrollment.program_uuid,
|
||||
'role': role_assignment.role,
|
||||
'user_id': user.id,
|
||||
'user_key': program_enrollment.external_user_key,
|
||||
}
|
||||
|
||||
try:
|
||||
with transaction.atomic():
|
||||
logger.info('Creating access role %(role)s for user with user id %(user_id)s and '
|
||||
'external user key %(user_key)s for course with course key %(course_key)s '
|
||||
'in program with uuid %(program_uuid)s.',
|
||||
logger_format_values
|
||||
)
|
||||
# if the user already has the role, then the add users method ignores this
|
||||
# and the operation is a no-op
|
||||
role.add_users(user)
|
||||
# because the user now has a corresponding CourseAccessRole, we no longer need
|
||||
# the CourseAccessRoleAssignment object
|
||||
role_assignment.delete()
|
||||
except Exception: # pylint: disable=broad-except
|
||||
logger.error('Unable to create access role %(role)s for user with user id %(user_id)s and '
|
||||
'external user key %(user_key)s for course with course key %(course_key)s '
|
||||
'in program with uuid %(program_uuid)s or to delete the CourseAccessRoleAssignment '
|
||||
'with role %(role)s and ProgramCourseEnrollment %(program_course_enrollment)r.',
|
||||
logger_format_values
|
||||
)
|
||||
|
||||
@@ -10,9 +10,15 @@ from edx_django_utils.cache import RequestCache
|
||||
from opaque_keys.edx.keys import CourseKey
|
||||
from testfixtures import LogCapture
|
||||
|
||||
from lms.djangoapps.program_enrollments.tests.factories import ProgramCourseEnrollmentFactory, ProgramEnrollmentFactory
|
||||
from lms.djangoapps.program_enrollments.tests.factories import (
|
||||
CourseAccessRoleAssignmentFactory,
|
||||
ProgramCourseEnrollmentFactory,
|
||||
ProgramEnrollmentFactory,
|
||||
)
|
||||
from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
|
||||
from student.tests.factories import UserFactory
|
||||
from student.api import get_course_access_role
|
||||
from student.roles import CourseStaffRole
|
||||
from student.tests.factories import UserFactory, CourseAccessRoleFactory
|
||||
|
||||
from ..linking import (
|
||||
NO_LMS_USER_TEMPLATE,
|
||||
@@ -179,6 +185,77 @@ class TestLinkProgramEnrollments(TestLinkProgramEnrollmentsMixin, TestCase):
|
||||
self.assertEqual(inactive_enrollment.course_enrollment.course.id, self.animal_course)
|
||||
self.assertFalse(inactive_enrollment.course_enrollment.is_active)
|
||||
|
||||
def test_realize_course_access_roles(self):
|
||||
program_enrollment = self._create_waiting_enrollment(self.program, '0001')
|
||||
active_enrollment_1 = self._create_waiting_course_enrollment(
|
||||
program_enrollment,
|
||||
self.fruit_course,
|
||||
status='active'
|
||||
)
|
||||
active_enrollment_2 = self._create_waiting_course_enrollment(
|
||||
program_enrollment,
|
||||
self.animal_course,
|
||||
status='active'
|
||||
)
|
||||
CourseAccessRoleAssignmentFactory(enrollment=active_enrollment_1)
|
||||
CourseAccessRoleAssignmentFactory(enrollment=active_enrollment_2)
|
||||
link_program_enrollments(self.program, {'0001': self.user_1.username})
|
||||
|
||||
# assert that staff CourseAccessRoles are created for the user in the courses
|
||||
fruit_course_staff_role = get_course_access_role(
|
||||
self.user_1,
|
||||
self.fruit_course.org,
|
||||
self.fruit_course,
|
||||
CourseStaffRole.ROLE
|
||||
)
|
||||
assert fruit_course_staff_role is not None
|
||||
|
||||
animal_course_staff_role = get_course_access_role(
|
||||
self.user_1,
|
||||
self.animal_course.org,
|
||||
self.animal_course,
|
||||
CourseStaffRole.ROLE
|
||||
)
|
||||
assert animal_course_staff_role is not None
|
||||
|
||||
# assert that all CourseAccessRoleAssignment objects are deleted
|
||||
assert not active_enrollment_1.courseaccessroleassignment_set.all().exists()
|
||||
assert not active_enrollment_2.courseaccessroleassignment_set.all().exists()
|
||||
|
||||
def test_realize_course_access_roles_user_with_existing_course_access_role(self):
|
||||
"""
|
||||
This test asserts that, given a user that already has a staff CourseAccessRole in a course,
|
||||
if that user has a CourseAccessRoleAssignment that describes a staff role in that same course,
|
||||
that we do not mistakenly violate the unique_together constraint on the CourseAccessRole model by
|
||||
creating a duplicate. As of now, this is handled by the CourseStaffRole code itself, which silently
|
||||
ignores such duplicates, but this test is to ensure we do not regress.
|
||||
"""
|
||||
program_enrollment = self._create_waiting_enrollment(self.program, '0001')
|
||||
active_enrollment_1 = self._create_waiting_course_enrollment(
|
||||
program_enrollment,
|
||||
self.fruit_course,
|
||||
status='active'
|
||||
)
|
||||
# create an CourseAccessRole for the user
|
||||
CourseAccessRoleFactory(user=self.user_1, course_id=self.fruit_course, role=CourseStaffRole.ROLE)
|
||||
|
||||
# create a corresponding CourseAccessRoleAssignmentFactory that would, theoretically, cause a
|
||||
# duplicate object to be created, violating the CourseAccessRole integrity constraints
|
||||
CourseAccessRoleAssignmentFactory(enrollment=active_enrollment_1)
|
||||
link_program_enrollments(self.program, {'0001': self.user_1.username})
|
||||
|
||||
# assert that staff CourseAccessRoles remains
|
||||
fruit_course_staff_role = get_course_access_role(
|
||||
self.user_1,
|
||||
self.fruit_course.org,
|
||||
self.fruit_course,
|
||||
CourseStaffRole.ROLE
|
||||
)
|
||||
assert fruit_course_staff_role is not None
|
||||
|
||||
# assert that all CourseAccessRoleAssignment objects are deleted
|
||||
assert not active_enrollment_1.courseaccessroleassignment_set.all().exists()
|
||||
|
||||
|
||||
class TestLinkProgramEnrollmentsErrors(TestLinkProgramEnrollmentsMixin, TestCase):
|
||||
""" Tests for linking error behavior """
|
||||
|
||||
Reference in New Issue
Block a user