Skip to content

E
edx-platform
Commit babd972c by Matt Drayer Committed by Jonathan Piacenti
Options

mattdrayer/api-course-exists-check: Refactored course lookup logic

parent 7cab70b6
Showing with 140 additions and 101 deletions
lms/djangoapps/api_manager/courses/tests.py
......@@ -474,7 +474,7 @@ class CoursesApiTests(TestCase):
def test_courses_groups_list_get(self):
test_uri = '{}/{}/groups'.format(self.base_courses_uri, self.test_course_id)
course_fail_uri = '{}/{}/groups'.format(self.base_courses_uri, '/ed/Open_DemoX/edx_demo_course')
course_fail_uri = '{}/{}/groups'.format(self.base_courses_uri, 'ed/Open_DemoX/edx_demo_course')
for i in xrange(2):
data_dict = {
'name': 'Alpha Group {}'.format(i), 'type': 'Programming',
......@@ -1311,7 +1311,6 @@ class CoursesApiTests(TestCase):
data = {'group_id': group_id}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 201)
# Create another group and add it to course module
data = {'name': 'Beta Group', 'type': 'project'}
response = self.do_post(self.base_groups_uri, data)
......@@ -1320,7 +1319,6 @@ class CoursesApiTests(TestCase):
data = {'group_id': another_group_id}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 201)
# create a 5 new users
for i in xrange(1, 6):
data = {
......@@ -1344,12 +1342,10 @@ class CoursesApiTests(TestCase):
data = {'user_id': created_user_id}
response = self.do_post(test_group_users_uri, data)
self.assertEqual(response.status_code, 201)
#enroll one user in Alpha Group and one in Beta Group created user
if i >= 2:
response = self.do_post(test_course_users_uri, data)
self.assertEqual(response.status_code, 201)
response = self.do_get('{}?enrolled={}'.format(test_uri_users, 'True'))
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data), 2)
......
lms/djangoapps/api_manager/courseware_access.py
......@@ -10,64 +10,32 @@ from xmodule.modulestore.django import modulestore
from xmodule.modulestore.exceptions import ItemNotFoundError
def get_course(request, user, course_id, depth=0):
def get_course(request, user, course_id, depth=0, load_content=False):
"""
Utility method to obtain course components
"""
course_descriptor = None
course_key = None
course_content = None
try:
course_key = CourseKey.from_string(course_id)
except InvalidKeyError:
try:
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_id)
except InvalidKeyError:
pass
course_key = get_course_key(course_id)
if course_key:
try:
course_descriptor = courses.get_course(course_key, depth)
except ValueError:
pass
if course_descriptor:
field_data_cache = FieldDataCache([course_descriptor], course_key, user)
course_content = module_render.get_module_for_descriptor(
user,
request,
course_descriptor,
field_data_cache,
course_key)
course_descriptor = get_course_descriptor(course_key, depth)
if course_descriptor and load_content:
course_content = get_course_content(request, user, course_key, course_descriptor)
return course_descriptor, course_key, course_content
def get_course_child(request, user, course_key, content_id):
def get_course_child(request, user, course_key, content_id, load_content=False):
"""
Return a course xmodule/xblock to the caller
"""
content_descriptor = None
content_key = None
content = None
try:
content_key = UsageKey.from_string(content_id)
except InvalidKeyError:
try:
content_key = Location.from_deprecated_string(content_id)
except (InvalidLocationError, InvalidKeyError):
pass
if content_key:
try:
content_descriptor = modulestore().get_item(content_key)
except ItemNotFoundError:
pass
if content_descriptor:
field_data_cache = FieldDataCache([content_descriptor], course_key, user)
content = module_render.get_module_for_descriptor(
user,
request,
content_descriptor,
field_data_cache,
course_key)
return content_descriptor, content_key, content
child_descriptor = None
child_content = None
child_key = get_course_child_key(content_id)
if child_key:
child_descriptor = get_course_child_descriptor(child_key)
if child_descriptor and load_content:
child_content = get_course_child_content(request, user, course_key, child_descriptor)
return child_descriptor, child_key, child_content
def get_course_total_score(course_summary):
......@@ -92,3 +60,72 @@ def get_course_leaf_nodes(course_key, detached_categories):
nodes.extend([unit.location for unit in vertical.get_children()
if getattr(unit, 'category') not in detached_categories])
return nodes
def get_course_key(course_id):
try:
course_key = CourseKey.from_string(course_id)
except InvalidKeyError:
try:
course_key = SlashSeparatedCourseKey.from_deprecated_string(course_id)
except InvalidKeyError:
course_key = None
return course_key
def get_course_descriptor(course_key, depth):
try:
course_descriptor = courses.get_course(course_key, depth)
except ValueError:
course_descriptor = None
return course_descriptor
def get_course_content(request, user, course_key, course_descriptor):
field_data_cache = FieldDataCache([course_descriptor], course_key, user)
course_content = module_render.get_module_for_descriptor(
user,
request,
course_descriptor,
field_data_cache,
course_key)
return course_content
def course_exists(request, user, course_id):
course_key = get_course_key(course_id)
if not course_key:
return False
if not modulestore().has_course(course_key):
return False
return True
def get_course_child_key(content_id):
try:
content_key = UsageKey.from_string(content_id)
except InvalidKeyError:
try:
content_key = Location.from_deprecated_string(content_id)
except (InvalidLocationError, InvalidKeyError):
content_key = None
return content_key
def get_course_child_descriptor(child_key):
try:
content_descriptor = modulestore().get_item(child_key)
except ItemNotFoundError:
content_descriptor = None
return content_descriptor
def get_course_child_content(request, user, course_key, child_descriptor):
field_data_cache = FieldDataCache([child_descriptor], course_key, user)
child_content = module_render.get_module_for_descriptor(
user,
request,
child_descriptor,
field_data_cache,
course_key)
return child_content
lms/djangoapps/api_manager/users/tests.py
......@@ -22,7 +22,7 @@ from capa.tests.response_xml_factory import StringResponseXMLFactory
from courseware.tests.factories import StudentModuleFactory
from django_comment_common.models import Role, FORUM_ROLE_MODERATOR
from instructor.access import allow_access
from projects.models import Project
from projects.models import Project, Workgroup
from student.tests.factories import UserFactory
from student.models import anonymous_id_for_user
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
......@@ -60,6 +60,7 @@ class UsersApiTests(ModuleStoreTestCase):
self.groups_base_uri = '/api/server/groups'
self.org_base_uri = '/api/server/organizations/'
self.workgroups_base_uri = '/api/server/workgroups/'
self.projects_base_uri = '/api/server/projects/'
self.users_base_uri = '/api/server/users'
self.sessions_base_uri = '/api/server/sessions'
self.test_bogus_course_id = 'foo/bar/baz'
......@@ -1251,47 +1252,46 @@ class UsersApiTests(ModuleStoreTestCase):
def test_user_workgroups_list(self):
test_workgroups_uri = self.workgroups_base_uri
user_id = self.user.id
# create anonymous user
anonymous_id = anonymous_id_for_user(self.user, self.course.id)
for i in xrange(1, 12):
course = CourseFactory.create(
display_name="TEST COURSE {}".format(i),
)
course_content = ItemFactory.create(
category="videosequence",
parent_location=course.location,
data=self.test_course_data,
display_name="View_Sequence"
)
project_1 = Project.objects.create(
course_id=unicode(self.course.id),
content_id=unicode(self.course_content.scope_ids.usage_id),
)
p1_workgroup_1 = Workgroup.objects.create(
name = 'Workgroup 1',
project = project_1
)
test_project = Project.objects.create(
course_id=unicode(course.id),
content_id=unicode(course_content.scope_ids.usage_id)
)
data = {
'name': 'Workgroup ' + str(i),
'project': test_project.id
}
response = self.do_post(test_workgroups_uri, data)
self.assertEqual(response.status_code, 201)
test_uri = '{}{}/'.format(test_workgroups_uri, str(response.data['id']))
users_uri = '{}users/'.format(test_uri)
data = {"id": user_id}
project_2 = Project.objects.create(
course_id=unicode(self.course2.id),
content_id=unicode(self.course2_content.scope_ids.usage_id),
)
p2_workgroup_1 = Workgroup.objects.create(
name = 'Workgroup 2',
project = project_2
)
for i in xrange(1,12):
test_user = UserFactory()
users_uri = '{}{}/users/'.format(self.workgroups_base_uri, 1)
data = {"id": test_user.id}
response = self.do_post(users_uri, data)
self.assertEqual(response.status_code, 201)
if test_user.id > 6:
users_uri = '{}{}/users/'.format(self.workgroups_base_uri, 2)
data = {"id": test_user.id}
response = self.do_post(users_uri, data)
self.assertEqual(response.status_code, 201)
# test with anonymous user id
test_uri = '{}/{}/workgroups/?page_size=10'.format(self.users_base_uri, anonymous_id)
anonymous_id = anonymous_id_for_user(test_user, self.course.id)
test_uri = '{}/{}/workgroups/?page_size=1'.format(self.users_base_uri, anonymous_id)
response = self.do_get(test_uri)
self.assertEqual(response.data['count'], 11)
self.assertEqual(len(response.data['results']), 10)
self.assertEqual(response.data['count'], 2)
self.assertEqual(len(response.data['results']), 1)
self.assertEqual(response.data['num_pages'], 2)
# test with course_id filter and integer user id
course_id = {'course_id': unicode(course.id)}
response = self.do_get('{}/{}/workgroups/?{}'.format(self.users_base_uri, user_id, urlencode(course_id)))
course_id = {'course_id': unicode(self.course.id)}
response = self.do_get('{}/{}/workgroups/?{}'.format(self.users_base_uri, test_user.id, urlencode(course_id)))
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data['count'], 1)
self.assertEqual(len(response.data['results']), 1)
......
lms/djangoapps/api_manager/users/views.py
......@@ -34,7 +34,7 @@ from util.password_policy_validators import (
)
from api_manager.courses.serializers import CourseModuleCompletionSerializer
from api_manager.courseware_access import get_course, get_course_child, get_course_total_score
from api_manager.courseware_access import get_course, get_course_child, get_course_total_score, get_course_key, course_exists
from api_manager.permissions import SecureAPIView, SecureListAPIView, IdsInFilterBackend, HasOrgsFilterBackend
from api_manager.models import GroupProfile, APIUser as User
from api_manager.organizations.serializers import OrganizationSerializer
......@@ -90,10 +90,9 @@ def _save_content_position(request, user, course_key, position):
parent_content_id = position['parent_content_id']
child_content_id = position['child_content_id']
if unicode(course_key) == parent_content_id:
parent_descriptor, parent_key, parent_content = get_course(request, user, parent_content_id) # pylint: disable=W0612
parent_descriptor, parent_key, parent_content = get_course(request, user, parent_content_id, load_content=True) # pylint: disable=W0612
else:
parent_descriptor, parent_key, parent_content = get_course_child(request, user, course_key, parent_content_id) # pylint: disable=W0612
parent_descriptor, parent_key, parent_content = get_course_child(request, user, course_key, parent_content_id, load_content=True) # pylint: disable=W0612
if not parent_descriptor:
return None
......@@ -126,8 +125,7 @@ def _save_child_position(parent_descriptor, target_child_location):
# Only save if position changed
if position != parent_descriptor.position:
parent_descriptor.position = position
# Save this new position to the underlying KeyValueStore
parent_descriptor.save()
parent_descriptor.save()
def _manage_role(course_descriptor, user, role, action):
......@@ -807,13 +805,13 @@ class UsersCoursesDetail(SecureAPIView):
user = User.objects.get(id=user_id, is_active=True)
except ObjectDoesNotExist:
return Response({}, status=status.HTTP_404_NOT_FOUND)
course_descriptor, course_key, course_content = get_course(request, user, course_id) # pylint: disable=W0612
if not course_descriptor:
if not course_exists(request, user, course_id):
return Response({}, status=status.HTTP_404_NOT_FOUND)
response_data['user_id'] = user.id
response_data['course_id'] = course_id
if request.DATA['positions']:
course_key = get_course_key(course_id)
response_data['positions'] = []
for position in request.DATA['positions']:
content_position = _save_content_position(
......@@ -865,7 +863,7 @@ class UsersCoursesDetail(SecureAPIView):
response_data['position_tree'][current_child_loc.category] = {}
response_data['position_tree'][current_child_loc.category]['id'] = unicode(current_child_loc)
_,_,parent_module = get_course_child(request, user, course_key, unicode(current_child_loc))
_,_,parent_module = get_course_child(request, user, course_key, unicode(current_child_loc), load_content=True)
else:
parent_module = None
......@@ -879,9 +877,9 @@ class UsersCoursesDetail(SecureAPIView):
user = User.objects.get(id=user_id, is_active=True)
except ObjectDoesNotExist:
return Response({}, status=status.HTTP_204_NO_CONTENT)
course_descriptor, course_key, course_content = get_course(request, user, course_id) # pylint: disable=W0612
if not course_descriptor:
if not course_exists(request, user, course_id):
return Response({}, status=status.HTTP_204_NO_CONTENT)
course_key = get_course_key(course_id)
CourseEnrollment.unenroll(user, course_key)
return Response({}, status=status.HTTP_204_NO_CONTENT)
......@@ -1234,9 +1232,9 @@ class UsersRolesList(SecureListAPIView):
course_id = self.request.QUERY_PARAMS.get('course_id', None)
if course_id:
course_descriptor, course_key, course_content = get_course(self.request, user, course_id) # pylint: disable=W0612
if not course_descriptor:
if not course_exists(self.request, user, course_id):
raise Http404
course_key = get_course_key(course_id)
queryset = queryset.filter(course_id=course_key)
role = self.request.QUERY_PARAMS.get('role', None)
......
lms/djangoapps/projects/views.py
......@@ -30,7 +30,7 @@ from .serializers import ProjectSerializer, WorkgroupSerializer, WorkgroupSubmis
from .serializers import WorkgroupReviewSerializer, WorkgroupSubmissionReviewSerializer, WorkgroupPeerReviewSerializer
def _get_course(request, user, course_id, depth=0):
def _get_course(request, user, course_id, depth=0, load_content=False):
"""
Utility method to obtain course components
"""
......@@ -49,7 +49,7 @@ def _get_course(request, user, course_id, depth=0):
course_descriptor = get_course(course_key, depth=depth)
except ValueError:
pass
if course_descriptor:
if course_descriptor and load_content:
field_data_cache = FieldDataCache([course_descriptor], course_key, user)
course_content = module_render.get_module(
user,
......@@ -60,7 +60,7 @@ def _get_course(request, user, course_id, depth=0):
return course_descriptor, course_key, course_content
def _get_course_child(request, user, course_key, content_id):
def _get_course_child(request, user, course_key, content_id, load_content=False):
"""
Return a course xmodule/xblock to the caller
"""
......@@ -77,7 +77,7 @@ def _get_course_child(request, user, course_key, content_id):
if content_key:
store = modulestore()
content_descriptor = store.get_item(content_key)
if content_descriptor:
if content_descriptor and load_content:
field_data_cache = FieldDataCache([content_descriptor], course_key, user)
content = module_render.get_module(
user,
......
lms/envs/test.py
......@@ -500,3 +500,11 @@ PROFILE_IMAGE_MIN_BYTES = 100
FEATURES['ENABLE_LTI_PROVIDER'] = True
INSTALLED_APPS += ('lti_provider',)
AUTHENTICATION_BACKENDS += ('lti_provider.users.LtiBackend',)
########################## SECURITY #######################
FEATURES['ENFORCE_PASSWORD_POLICY'] = False
FEATURES['ENABLE_MAX_FAILED_LOGIN_ATTEMPTS'] = False
FEATURES['SQUELCH_PII_IN_LOGS'] = False
FEATURES['PREVENT_CONCURRENT_LOGINS'] = False
FEATURES['ADVANCED_SECURITY'] = False
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment