Skip to content

E
edx-platform
Commit 98793cb7 by Martyn James
Options

Merge pull request #28 from edx-solutions/mattdrayer/api-cbv 

Migrated from Function-Based Views to Class-Based-Views
parents fa6aaf8e e5656661
Showing with 1509 additions and 998 deletions
lms/djangoapps/api_manager/courses_urls.py
......@@ -4,19 +4,24 @@ The order of the URIs really matters here, due to the slash characters present i
"""
from django.conf.urls import patterns, url
urlpatterns = patterns(
'api_manager.courses_views',
url(r'/*$^', 'courses_list'),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/modules/(?P<module_id>[a-zA-Z0-9/_:]+)/submodules/*$', 'modules_list'),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/modules/(?P<module_id>[a-zA-Z0-9/_:]+)$', 'modules_detail'),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/modules/*$', 'modules_list'),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/groups/(?P<group_id>[0-9]+)$', 'courses_groups_detail'),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/groups/*$', 'courses_groups_list'),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/overview$', 'course_overview'),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/updates$', 'course_updates'),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/static_tabs/(?P<tab_id>[a-zA-Z0-9/_:]+)$', 'static_tab_detail'),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/static_tabs$', 'static_tabs_list'),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/users$', 'course_users_list'),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/tree/(?P<depth>[0-9]+)$', 'course_tree'),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)$', 'courses_detail'),
from rest_framework.urlpatterns import format_suffix_patterns
from api_manager import courses_views
urlpatterns = patterns('',
url(r'/*$^', courses_views.CoursesList.as_view()),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)$', courses_views.CoursesDetail.as_view()),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/modules/(?P<module_id>[a-zA-Z0-9/_:]+)/submodules/*$', courses_views.ModulesList.as_view()),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/modules/(?P<module_id>[a-zA-Z0-9/_:]+)$', courses_views.ModulesDetail.as_view()),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/modules/*$', courses_views.ModulesList.as_view()),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/groups/(?P<group_id>[0-9]+)$', courses_views.CoursesGroupsDetail.as_view()),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/groups/*$', courses_views.CoursesGroupsList.as_view()),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/overview$', courses_views.CoursesOverview.as_view()),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/updates$', courses_views.CoursesUpdates.as_view()),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/static_tabs/(?P<tab_id>[a-zA-Z0-9/_:]+)$', courses_views.CoursesStaticTabsDetail.as_view()),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/static_tabs$', courses_views.CoursesStaticTabsList.as_view()),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/users/(?P<user_id>[0-9]+)$', courses_views.CoursesUsersDetail.as_view()),
url(r'^(?P<course_id>[^/]+/[^/]+/[^/]+)/users$', courses_views.CoursesUsersList.as_view()),
)
urlpatterns = format_suffix_patterns(urlpatterns)
lms/djangoapps/api_manager/courses_views.py
""" API implementation for course-oriented interactions. """
from django.contrib.auth.models import Group, User
from django.core.exceptions import ObjectDoesNotExist
from lxml import etree
from StringIO import StringIO
from collections import OrderedDict
import logging
from lxml import etree
from StringIO import StringIO
from django.contrib.auth.models import Group, User
from django.core.exceptions import ObjectDoesNotExist
from django.http import Http404
from rest_framework import status
from rest_framework.decorators import api_view, permission_classes
from rest_framework.response import Response
from rest_framework.views import APIView
from api_manager.permissions import ApiKeyHeaderPermission
from api_manager.models import CourseGroupRelationship
from courseware import module_render
from courseware.courses import get_course, get_course_about_section, get_course_info_section
from courseware.model_data import FieldDataCache
from courseware.views import get_static_tab_contents
from student.models import CourseEnrollment, CourseEnrollmentAllowed
from xmodule.modulestore.django import modulestore
from xmodule.modulestore import Location, InvalidLocationError
from courseware.courses import get_course_about_section, get_course_info_section, get_course_by_id
from courseware.views import get_static_tab_contents
from student.models import CourseEnrollment, CourseEnrollmentAllowed
log = logging.getLogger(__name__)
......@@ -32,7 +37,7 @@ def _generate_base_uri(request):
resource_uri = '{}://{}{}'.format(
protocol,
request.get_host(),
request.path
request.get_full_path()
)
return resource_uri
......@@ -111,233 +116,22 @@ def _serialize_module_submodules(request, course_id, submodules):
return data
@api_view(['GET'])
@permission_classes((ApiKeyHeaderPermission,))
def modules_list(request, course_id, module_id=None):
"""
GET retrieves the list of submodules for a given module
We don't know where in the module hierarchy we are -- could even be the top
"""
if module_id is None:
module_id = course_id
response_data = []
submodule_type = request.QUERY_PARAMS.get('type', None)
store = modulestore()
if course_id != module_id:
try:
module = store.get_instance(course_id, Location(module_id))
except InvalidLocationError:
module = None
else:
module = store.get_course(course_id)
if module:
submodules = _get_module_submodules(module, submodule_type)
response_data = _serialize_module_submodules(
request,
course_id,
submodules
)
status_code = status.HTTP_200_OK
else:
status_code = status.HTTP_404_NOT_FOUND
return Response(response_data, status=status_code)
@api_view(['GET'])
@permission_classes((ApiKeyHeaderPermission,))
def modules_detail(request, course_id, module_id):
"""
GET retrieves an existing module from the system
"""
store = modulestore()
response_data = {}
submodule_type = request.QUERY_PARAMS.get('type', None)
if course_id != module_id:
try:
module = store.get_instance(course_id, Location(module_id))
except InvalidLocationError:
module = None
else:
module = store.get_course(course_id)
if module:
response_data = _serialize_module(
request,
course_id,
module
)
submodules = _get_module_submodules(module, submodule_type)
response_data['modules'] = _serialize_module_submodules(
request,
course_id,
submodules
)
status_code = status.HTTP_200_OK
else:
status_code = status.HTTP_404_NOT_FOUND
return Response(response_data, status=status_code)
@api_view(['GET'])
@permission_classes((ApiKeyHeaderPermission,))
def courses_list(request):
"""
GET returns the list of available courses
"""
response_data = []
store = modulestore()
course_descriptors = store.get_courses()
for course_descriptor in course_descriptors:
course_data = _serialize_module(
request,
course_descriptor.id,
course_descriptor
)
response_data.append(course_data)
return Response(response_data, status=status.HTTP_200_OK)
@api_view(['GET'])
@permission_classes((ApiKeyHeaderPermission,))
def courses_detail(request, course_id):
"""
GET retrieves an existing course from the system
"""
response_data = {}
store = modulestore()
try:
course_descriptor = store.get_course(course_id)
except ValueError:
course_descriptor = None
if course_descriptor:
response_data = _serialize_module(
request,
course_descriptor.id,
course_descriptor
)
submodules = _get_module_submodules(course_descriptor, None)
response_data['modules'] = _serialize_module_submodules(
request,
course_id,
submodules
)
status_code = status.HTTP_200_OK
else:
status_code = status.HTTP_404_NOT_FOUND
return Response(response_data, status=status_code)
@api_view(['GET'])
@permission_classes((ApiKeyHeaderPermission,))
def course_tree(request, course_id, depth):
"""
GET retrieves an existing course from the system and returns summary information about the submodules
to the specified depth
"""
response_data = {}
depth_int = int(depth)
# note, passing in depth=N optimizes the number of round trips to the database
course_descriptor = get_course_by_id(course_id, depth=depth_int)
if not course_descriptor:
return Response({}, status.HTTP_404_NOT_FOUND)
def _serialize_node_with_children(descriptor, depth):
data = _serialize_module(
request,
course_descriptor.id,
descriptor
)
if depth > 0:
data['modules'] = []
for child in descriptor.get_children():
data['modules'].append(_serialize_node_with_children(child, depth-1))
return data
response_data = _serialize_node_with_children(course_descriptor, depth_int)
return Response(response_data)
@api_view(['POST'])
@permission_classes((ApiKeyHeaderPermission,))
def courses_groups_list(request, course_id):
"""
POST creates a new course-group relationship in the system
"""
response_data = {}
group_id = request.DATA['group_id']
base_uri = _generate_base_uri(request)
store = modulestore()
try:
existing_course = store.get_course(course_id)
except ValueError:
existing_course = None
try:
existing_group = Group.objects.get(id=group_id)
except ObjectDoesNotExist:
existing_group = None
if existing_course and existing_group:
try:
existing_relationship = CourseGroupRelationship.objects.get(course_id=course_id, group=existing_group)
except ObjectDoesNotExist:
existing_relationship = None
if existing_relationship is None:
CourseGroupRelationship.objects.create(course_id=course_id, group=existing_group)
response_data['course_id'] = str(existing_course.id)
response_data['group_id'] = str(existing_group.id)
response_data['uri'] = '{}/{}'.format(base_uri, existing_group.id)
response_status = status.HTTP_201_CREATED
else:
response_data['message'] = "Relationship already exists."
response_status = status.HTTP_409_CONFLICT
else:
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
@api_view(['GET', 'DELETE'])
@permission_classes((ApiKeyHeaderPermission,))
def courses_groups_detail(request, course_id, group_id):
"""
GET retrieves an existing course-group relationship from the system
DELETE removes/inactivates/etc. an existing course-group relationship
"""
if request.method == 'GET':
response_data = {}
base_uri = _generate_base_uri(request)
response_data['uri'] = base_uri
response_data['course_id'] = course_id
response_data['group_id'] = group_id
store = modulestore()
try:
existing_course = store.get_course(course_id)
except ValueError:
existing_course = None
try:
existing_group = Group.objects.get(id=group_id)
except ObjectDoesNotExist:
existing_group = None
if existing_course and existing_group:
try:
existing_relationship = CourseGroupRelationship.objects.get(course_id=course_id, group=existing_group)
except ObjectDoesNotExist:
existing_relationship = None
if existing_relationship:
response_status = status.HTTP_200_OK
else:
response_status = status.HTTP_404_NOT_FOUND
else:
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
elif request.method == 'DELETE':
try:
existing_group = Group.objects.get(id=group_id)
existing_relationship = CourseGroupRelationship.objects.get(course_id=course_id, group=existing_group).delete()
except ObjectDoesNotExist:
pass
return Response({}, status=status.HTTP_204_NO_CONTENT)
def _serialize_module_with_children(request, course_descriptor, descriptor, depth):
data = _serialize_module(
request,
course_descriptor.id,
descriptor
)
if depth > 0:
data['modules'] = []
for child in descriptor.get_children():
data['modules'].append(_serialize_module_with_children(
request,
course_descriptor,
child,
depth-1
))
return data
def _inner_content(tag):
......@@ -409,41 +203,6 @@ def _parse_overview_html(html):
return result
@api_view(['GET'])
@permission_classes((ApiKeyHeaderPermission,))
def course_overview(request, course_id):
"""
GET retrieves the course overview module, which - in MongoDB - is stored with the following
naming convention {"_id.org":"i4x", "_id.course":<course_num>, "_id.category":"about", "_id.name":"overview"}
"""
store = modulestore()
response_data = OrderedDict()
try:
course_module = store.get_course(course_id)
if not course_module:
return Response({}, status=status.HTTP_404_NOT_FOUND)
content = get_course_about_section(course_module, 'overview')
if request.GET.get('parse') and request.GET.get('parse') in ['True', 'true']:
try:
response_data['sections'] = _parse_overview_html(content)
except:
log.exception(
u"Error prasing course overview. Content = {0}".format(
content
))
return Response({'err': 'could_not_parse'}, status=status.HTTP_409_CONFLICT)
else:
response_data['overview_html'] = content
except InvalidLocationError:
return Response({}, status=status.HTTP_404_NOT_FOUND)
return Response(response_data)
def _parse_updates_html(html):
"""
Helper method to break up the course updates HTML into components
......@@ -478,133 +237,383 @@ def _parse_updates_html(html):
return result
@api_view(['GET'])
@permission_classes((ApiKeyHeaderPermission,))
def course_updates(request, course_id):
"""
GET retrieves the course overview module, which - in MongoDB - is stored with the following
naming convention {"_id.org":"i4x", "_id.course":<course_num>, "_id.category":"course_info", "_id.name":"updates"}
"""
store = modulestore()
response_data = OrderedDict()
class ModulesList(APIView):
permission_classes = (ApiKeyHeaderPermission,)
try:
course_module = store.get_course(course_id)
if not course_module:
return Response({}, status=status.HTTP_404_NOT_FOUND)
def get(self, request, course_id, module_id=None, format=None):
"""
GET retrieves the list of submodules for a given module
We don't know where in the module hierarchy we are -- could even be the top
"""
if module_id is None:
module_id = course_id
response_data = []
submodule_type = request.QUERY_PARAMS.get('type', None)
store = modulestore()
if course_id != module_id:
try:
module = store.get_instance(course_id, Location(module_id))
except InvalidLocationError:
module = None
else:
module = get_course(course_id)
if module:
submodules = _get_module_submodules(module, submodule_type)
response_data = _serialize_module_submodules(
request,
course_id,
submodules
)
status_code = status.HTTP_200_OK
else:
status_code = status.HTTP_404_NOT_FOUND
return Response(response_data, status=status_code)
content = get_course_info_section(request, course_module, 'updates')
if not content:
class ModulesDetail(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def get(self, request, course_id, module_id, format=None):
"""
GET retrieves an existing module from the system
"""
store = modulestore()
response_data = {}
submodule_type = request.QUERY_PARAMS.get('type', None)
if course_id != module_id:
try:
module = store.get_instance(course_id, Location(module_id))
except InvalidLocationError:
module = None
else:
module = get_course(course_id)
if module:
response_data = _serialize_module(
request,
course_id,
module
)
submodules = _get_module_submodules(module, submodule_type)
response_data['modules'] = _serialize_module_submodules(
request,
course_id,
submodules
)
status_code = status.HTTP_200_OK
else:
status_code = status.HTTP_404_NOT_FOUND
return Response(response_data, status=status_code)
class CoursesList(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def get(self, request, format=None):
"""
GET returns the list of available courses
"""
response_data = []
store = modulestore()
course_descriptors = store.get_courses()
for course_descriptor in course_descriptors:
course_data = _serialize_module(
request,
course_descriptor.id,
course_descriptor
)
response_data.append(course_data)
return Response(response_data, status=status.HTTP_200_OK)
class CoursesDetail(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def get(self, request, course_id, format=None):
"""
GET retrieves an existing course from the system and returns summary information about the submodules
to the specified depth
"""
depth = request.QUERY_PARAMS.get('depth', 0)
depth_int = int(depth)
# get_course_by_id raises an Http404 if the requested course is invalid
# Rather than catching it, we just let it bubble up
try:
course_descriptor = get_course(course_id)
except ValueError:
course_descriptor = None
if course_descriptor:
if depth_int > 0:
response_data = _serialize_module_with_children(
request,
course_descriptor,
course_descriptor, # Primer for recursive function
depth_int
)
else:
response_data = _serialize_module(
request,
course_descriptor.id,
course_descriptor
)
status_code = status.HTTP_200_OK
response_data['uri'] = _generate_base_uri(request)
return Response(response_data, status=status_code)
else:
return Response({}, status=status.HTTP_404_NOT_FOUND)
if request.GET.get('parse') and request.GET.get('parse') in ['True', 'true']:
class CoursesGroupsList(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def post(self, request, course_id, format=None):
"""
POST creates a new course-group relationship in the system
"""
response_data = {}
group_id = request.DATA['group_id']
base_uri = _generate_base_uri(request)
try:
existing_course = get_course(course_id)
except ValueError:
existing_course = None
try:
existing_group = Group.objects.get(id=group_id)
except ObjectDoesNotExist:
existing_group = None
if existing_course and existing_group:
try:
response_data['postings'] = _parse_updates_html(content)
except:
log.exception(
u"Error prasing course updates. Content = {0}".format(
content
))
return Response({'err': 'could_not_parse'}, status=status.HTTP_409_CONFLICT)
existing_relationship = CourseGroupRelationship.objects.get(course_id=course_id, group=existing_group)
except ObjectDoesNotExist:
existing_relationship = None
if existing_relationship is None:
CourseGroupRelationship.objects.create(course_id=course_id, group=existing_group)
response_data['course_id'] = str(existing_course.id)
response_data['group_id'] = str(existing_group.id)
response_data['uri'] = '{}/{}'.format(base_uri, existing_group.id)
response_status = status.HTTP_201_CREATED
else:
response_data['message'] = "Relationship already exists."
response_status = status.HTTP_409_CONFLICT
else:
response_data['content'] = content
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
class CoursesGroupsDetail(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def get(self, request, course_id, group_id, format=None):
"""
GET retrieves an existing course-group relationship from the system
"""
response_data = {}
base_uri = _generate_base_uri(request)
response_data['uri'] = base_uri
response_data['course_id'] = course_id
response_data['group_id'] = group_id
try:
existing_course = get_course(course_id)
except ValueError:
existing_course = None
try:
existing_group = Group.objects.get(id=group_id)
except ObjectDoesNotExist:
existing_group = None
if existing_course and existing_group:
try:
existing_relationship = CourseGroupRelationship.objects.get(course_id=course_id, group=existing_group)
except ObjectDoesNotExist:
existing_relationship = None
if existing_relationship:
response_status = status.HTTP_200_OK
else:
response_status = status.HTTP_404_NOT_FOUND
else:
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
except InvalidLocationError:
return Response({}, status=status.HTTP_404_NOT_FOUND)
def delete(self, request, course_id, group_id, format=None):
"""
DELETE removes/inactivates/etc. an existing course-group relationship
"""
try:
existing_group = Group.objects.get(id=group_id)
existing_relationship = CourseGroupRelationship.objects.get(course_id=course_id, group=existing_group).delete()
except ObjectDoesNotExist:
pass
return Response({}, status=status.HTTP_204_NO_CONTENT)
return Response(response_data)
class CoursesOverview(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def get(self, request, course_id, format=None):
"""
GET retrieves the course overview module, which - in MongoDB - is stored with the following
naming convention {"_id.org":"i4x", "_id.course":<course_num>, "_id.category":"about", "_id.name":"overview"}
"""
response_data = OrderedDict()
try:
existing_course = get_course(course_id)
except ValueError:
existing_course = None
if existing_course:
existing_content = get_course_about_section(existing_course, 'overview')
if existing_content:
if request.GET.get('parse') and request.GET.get('parse') in ['True', 'true']:
response_data['sections'] = _parse_overview_html(existing_content)
else:
response_data['overview_html'] = existing_content
return Response(response_data, status=status.HTTP_200_OK)
else:
return Response({}, status=status.HTTP_404_NOT_FOUND)
else:
return Response({}, status=status.HTTP_404_NOT_FOUND)
@api_view(['GET'])
@permission_classes((ApiKeyHeaderPermission,))
def static_tabs_list(request, course_id):
"""
GET returns an array of Static Tabs inside of a course
"""
store = modulestore()
response_data = OrderedDict()
try:
course_module = store.get_course(course_id)
if not course_module:
class CoursesUpdates(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def get(self, request, course_id, format=None):
"""
GET retrieves the course overview module, which - in MongoDB - is stored with the following
naming convention {"_id.org":"i4x", "_id.course":<course_num>, "_id.category":"course_info", "_id.name":"updates"}
"""
response_data = OrderedDict()
try:
existing_course = get_course(course_id)
except ValueError:
existing_course = None
if not existing_course:
return Response({}, status=status.HTTP_404_NOT_FOUND)
content = get_course_info_section(request, existing_course, 'updates')
if not content:
return Response({}, status=status.HTTP_404_NOT_FOUND)
if request.GET.get('parse') and request.GET.get('parse') in ['True', 'true']:
response_data['postings'] = _parse_updates_html(content)
else:
response_data['content'] = content
return Response(response_data)
class CoursesStaticTabsList(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def get(self, request, course_id):
"""
GET returns an array of Static Tabs inside of a course
"""
try:
existing_course = get_course(course_id)
except ValueError:
existing_course = None
if not existing_course:
return Response({}, status=status.HTTP_404_NOT_FOUND)
response_data = OrderedDict()
tabs = []
for tab in course_module.tabs:
for tab in existing_course.tabs:
if tab.type == 'static_tab':
tab_data = OrderedDict()
tab_data['id'] = tab.url_slug
tab_data['name'] = tab.name
if request.GET.get('detail') and request.GET.get('detail') in ['True', 'true']:
tab_data['content'] = get_static_tab_contents(request,
course_module,
tab_data['content'] = get_static_tab_contents(
request,
existing_course,
tab,
wrap_xmodule_display=False
)
tabs.append(tab_data)
response_data['tabs'] = tabs
except InvalidLocationError:
return Response({}, status=status.HTTP_404_NOT_FOUND)
return Response(response_data)
return Response(response_data)
@api_view(['GET'])
@permission_classes((ApiKeyHeaderPermission,))
def static_tab_detail(request, course_id, tab_id):
"""
GET returns an array of Static Tabs inside of a course
"""
store = modulestore()
response_data = OrderedDict()
class CoursesStaticTabsDetail(APIView):
permission_classes = (ApiKeyHeaderPermission,)
try:
course_module = store.get_course(course_id)
if not course_module:
def get(self, request, course_id, tab_id):
"""
GET returns the specified static tab for the specified course
"""
try:
existing_course = get_course(course_id)
except ValueError:
existing_course = None
if existing_course:
response_data = OrderedDict()
for tab in existing_course.tabs:
if tab.type == 'static_tab' and tab.url_slug == tab_id:
response_data['id'] = tab.url_slug
response_data['name'] = tab.name
response_data['content'] = get_static_tab_contents(
request,
existing_course,
tab,
wrap_xmodule_display=False
)
if not response_data:
return Response({}, status=status.HTTP_404_NOT_FOUND)
return Response(response_data, status=status.HTTP_200_OK)
else:
return Response({}, status=status.HTTP_404_NOT_FOUND)
for tab in course_module.tabs:
if tab.type == 'static_tab' and tab.url_slug == tab_id:
response_data['id'] = tab.url_slug
response_data['name'] = tab.name
response_data['content'] = get_static_tab_contents(request,
course_module,
tab,
wrap_xmodule_display=False
)
except InvalidLocationError:
return Response({}, status=status.HTTP_404_NOT_FOUND)
if not response_data:
return Response({}, status=status.HTTP_404_NOT_FOUND)
return Response(response_data)
class CoursesUsersList(APIView):
permission_classes = (ApiKeyHeaderPermission,)
@api_view(['GET', 'POST', 'DELETE'])
@permission_classes((ApiKeyHeaderPermission,))
def course_users_list(request, course_id):
"""
GET returns a list of users enrolled in the course_id
POST enrolls a student in the course. Note, this can be a user_id or just an email, in case
the user does not exist in the system
"""
store = modulestore()
response_data = OrderedDict()
try:
# find the course
course_module = store.get_course(course_id)
if not course_module:
def post(self, request, course_id, format=None):
"""
POST enrolls a student in the course. Note, this can be a user_id or
just an email, in case the user does not exist in the system
"""
response_data = OrderedDict()
try:
existing_course = get_course(course_id)
except ValueError:
existing_course = None
if not existing_course:
return Response({}, status=status.HTTP_404_NOT_FOUND)
except InvalidLocationError:
return Response({}, status=status.HTTP_404_NOT_FOUND)
if 'user_id' in request.DATA:
user_id = request.DATA['user_id']
try:
existing_user = User.objects.get(id=user_id)
except ObjectDoesNotExist:
existing_user = None
if existing_user:
CourseEnrollment.enroll(existing_user, course_id)
return Response({}, status=status.HTTP_201_CREATED)
else:
return Response({}, status=status.HTTP_404_NOT_FOUND)
elif 'email' in request.DATA:
try:
email = request.DATA['email']
existing_user = User.objects.get(email=email)
except ObjectDoesNotExist:
if request.DATA.get('allow_pending'):
# If the email doesn't exist we assume the student does not exist
# and the instructor is pre-enrolling them
# Store the pre-enrollment data in the CourseEnrollmentAllowed table
# NOTE: This logic really should live in CourseEnrollment.....
cea, _ = CourseEnrollmentAllowed.objects.get_or_create(course_id=course_id, email=email)
cea.auto_enroll = True
cea.save()
return Response({}, status.HTTP_201_CREATED)
else:
return Response({}, status.HTTP_400_BAD_REQUEST)
else:
return Response({}, status=status.HTTP_400_BAD_REQUEST)
if request.method == 'GET':
def get(self, request, course_id, format=None):
"""
GET returns a list of users enrolled in the course_id
"""
response_data = OrderedDict()
try:
existing_course = get_course(course_id)
except ValueError:
existing_course = None
if not existing_course:
return Response({}, status=status.HTTP_404_NOT_FOUND)
# Get a list of all enrolled students
users = CourseEnrollment.users_enrolled_in(course_id)
response_data['enrollments'] = []
......@@ -625,31 +634,58 @@ def course_users_list(request, course_id):
response_data['pending_enrollments'].append(cea.email)
return Response(response_data)
elif request.method == 'POST':
if 'user_id' in request.DATA:
user_id = request.DATA['user_id']
try:
existing_user = User.objects.get(id=user_id)
CourseEnrollment.enroll(existing_user, course_id)
except ObjectDoesNotExist:
return Response({'err': 'user_does_not_exist'}, status=status.HTTP_400_BAD_REQUEST)
elif 'email' in request.DATA:
# If caller passed in an email, then let's look up user by email address
# if it doesn't exist then we need to assume that the student does not exist
# in our database and that the instructor is pre-enrolling ment
email = request.DATA['email']
try:
existing_user = User.objects.get(email=email)
CourseEnrollment.enroll(existing_user, course_id)
except ObjectDoesNotExist:
if not request.DATA.get('allow_pending', False):
return Response({'err': 'user_does_not_exist'}, status=status.HTTP_400_BAD_REQUEST)
class CoursesUsersDetail(APIView):
permission_classes = (ApiKeyHeaderPermission,)
# In this case we can pre-enroll a non-existing student. This is what the
# CourseEnrollmentAllowed table is for
# NOTE: This logic really should live in CourseEnrollment.....
cea, _ = CourseEnrollmentAllowed.objects.get_or_create(course_id=course_id, email=email)
cea.auto_enroll = True
cea.save()
def get(self, request, course_id, user_id, format=None):
"""
GET identifies an ACTIVE course enrollment for the specified user
"""
base_uri = _generate_base_uri(request)
response_data = {
'course_id': course_id,
'user_id': user_id,
'uri': base_uri,
}
try:
course_descriptor = get_course(course_id)
except ValueError:
course_descriptor = None
if not course_descriptor:
return Response(response_data, status=status.HTTP_404_NOT_FOUND)
try:
user = User.objects.get(id=user_id, is_active=True)
except ObjectDoesNotExist:
user = None
if user and CourseEnrollment.is_enrolled(user, course_id):
field_data_cache = FieldDataCache([course_descriptor], course_id, user)
course_module = module_render.get_module(
user,
request,
course_descriptor.location,
field_data_cache,
course_id)
response_data['position'] = course_module.position
response_status = status.HTTP_200_OK
else:
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
return Response({}, status.HTTP_201_CREATED)
def delete(self, request, course_id, user_id, format=None):
"""
DELETE unenrolls the specified user from the specified course
"""
response_data = OrderedDict()
try:
existing_course = get_course(course_id)
except ValueError:
existing_course = None
if not existing_course:
return Response({}, status=status.HTTP_404_NOT_FOUND)
try:
user = User.objects.get(id=user_id, is_active=True)
except ObjectDoesNotExist:
user = None
if user:
CourseEnrollment.unenroll(user, course_id)
return Response({}, status=status.HTTP_204_NO_CONTENT)
lms/djangoapps/api_manager/groups_urls.py
""" Groups API URI specification """
from django.conf.urls import patterns, url
urlpatterns = patterns('api_manager.groups_views',
url(r'/*$^', 'group_list'),
url(r'^(?P<group_id>[0-9]+)$', 'group_detail'),
url(r'^(?P<group_id>[0-9]+)/courses/*$', 'group_courses_list'),
url(r'^(?P<group_id>[0-9]+)/courses/(?P<course_id>[a-zA-Z0-9/_:]+)$', 'group_courses_detail'),
url(r'^(?P<group_id>[0-9]+)/users/*$', 'group_users_list'),
url(r'^(?P<group_id>[0-9]+)/users/(?P<user_id>[0-9]+)$', 'group_users_detail'),
url(r'^(?P<group_id>[0-9]+)/groups/*$', 'group_groups_list'),
url(r'^(?P<group_id>[0-9]+)/groups/(?P<related_group_id>[0-9]+)$', 'group_groups_detail'),
)
from rest_framework.urlpatterns import format_suffix_patterns
from api_manager import groups_views
urlpatterns = patterns('',
url(r'/*$^', groups_views.GroupsList.as_view()),
url(r'^(?P<group_id>[0-9]+)$', groups_views.GroupsDetail.as_view()),
url(r'^(?P<group_id>[0-9]+)/courses/*$', groups_views.GroupsCoursesList.as_view()),
url(r'^(?P<group_id>[0-9]+)/courses/(?P<course_id>[a-zA-Z0-9/_:]+)$', groups_views.GroupsCoursesDetail.as_view()),
url(r'^(?P<group_id>[0-9]+)/users/*$', groups_views.GroupsUsersList.as_view()),
url(r'^(?P<group_id>[0-9]+)/users/(?P<user_id>[0-9]+)$', groups_views.GroupsUsersDetail.as_view()),
url(r'^(?P<group_id>[0-9]+)/groups/*$', groups_views.GroupsGroupsList.as_view()),
url(r'^(?P<group_id>[0-9]+)/groups/(?P<related_group_id>[0-9]+)$', groups_views.GroupsGroupsDetail.as_view()),
)
urlpatterns = format_suffix_patterns(urlpatterns)
lms/djangoapps/api_manager/groups_views.py
......@@ -10,6 +10,7 @@ from django.utils import timezone
from rest_framework import status
from rest_framework.decorators import api_view, permission_classes
from rest_framework.response import Response
from rest_framework.views import APIView
from api_manager.permissions import ApiKeyHeaderPermission
from api_manager.models import GroupRelationship, CourseGroupRelationship, GroupProfile
......@@ -29,36 +30,18 @@ def _generate_base_uri(request):
resource_uri = '{}://{}{}'.format(
protocol,
request.get_host(),
request.path
request.get_full_path()
)
return resource_uri
@api_view(['GET', 'POST'])
@permission_classes((ApiKeyHeaderPermission,))
def group_list(request):
"""
GET retrieves a list of groups in the system filtered by type
POST creates a new group in the system
"""
if request.method == 'GET':
if not 'type' in request.GET:
return Response({}, status=status.HTTP_400_BAD_REQUEST)
response_data = []
profiles = GroupProfile.objects.filter(group_type=request.GET['type'])
for profile in profiles:
item_data = {}
item_data['group_id'] = profile.group_id
if profile.group_type:
item_data['group_type'] = profile.group_type
if profile.data:
item_data['data'] = json.loads(profile.data)
response_data.append(item_data)
class GroupsList(APIView):
permissions_classes = (ApiKeyHeaderPermission,)
return Response(response_data)
elif request.method == 'POST':
def post(self, request, format=None):
"""
POST creates a new group in the system
"""
response_data = {}
base_uri = _generate_base_uri(request)
# Group name must be unique, but we need to support dupes
......@@ -71,118 +54,129 @@ def group_list(request):
group.record_date_modified = timezone.now()
group.save()
# Relationship model also allows us to use duplicate names
GroupRelationship.objects.create(name=original_group_name, group_id=group.id, parent_group=None)
# allow for optional meta information about groups, this will end up in the GroupProfile table
group_type = request.DATA.get('group_type')
data = json.dumps(request.DATA.get('data')) if request.DATA.get('data') else None
if group_type or data:
profile, _ = GroupProfile.objects.get_or_create(group_id=group.id, group_type=group_type, data=data)
response_data = {'id': group.id, 'name': original_group_name}
# Create a corresponding relationship management record
GroupRelationship.objects.create(group_id=group.id, parent_group=None)
# Create a corresponding profile record (for extra meta info)
group_type = request.DATA.get('group_type', None)
data = json.dumps(request.DATA.get('data')) if request.DATA.get('data') else {}
profile, _ = GroupProfile.objects.get_or_create(group_id=group.id, group_type=group_type, name=original_group_name, data=data)
response_data = {
'id': group.id,
'name': original_group_name,
'type': group_type,
}
base_uri = _generate_base_uri(request)
response_data['uri'] = '{}/{}'.format(base_uri, group.id)
response_status = status.HTTP_201_CREATED
return Response(response_data, status=response_status)
def get(self, request, format=None):
"""
GET retrieves a list of groups in the system filtered by type
"""
response_data = []
if 'type' in request.GET:
profiles = GroupProfile.objects.filter(group_type=request.GET['type'])
else:
profiles = GroupProfile.objects.all()
for profile in profiles:
item_data = {}
item_data['group_id'] = profile.group_id
if len(profile.name):
group_name = profile.name
else:
group = Group.objects.get(id=profile.group_id)
group_name = group.name
item_data['name'] = group_name
if profile.group_type:
item_data['group_type'] = profile.group_type
if profile.data:
item_data['data'] = json.loads(profile.data)
response_data.append(item_data)
return Response(response_data, status=status.HTTP_200_OK)
@api_view(['GET', 'POST'])
@permission_classes((ApiKeyHeaderPermission,))
def group_detail(request, group_id):
"""
GET retrieves an existing group from the system
"""
response_data = {}
base_uri = _generate_base_uri(request)
try:
existing_group = Group.objects.get(id=group_id)
existing_group_relationship = GroupRelationship.objects.get(group_id=group_id)
except ObjectDoesNotExist:
return Response({}, status.HTTP_404_NOT_FOUND)
if request.method == 'GET':
response_data['name'] = existing_group_relationship.name
response_data['id'] = existing_group.id
response_data['uri'] = base_uri
response_data['resources'] = []
resource_uri = '{}/users'.format(base_uri)
response_data['resources'].append({'uri': resource_uri})
resource_uri = '{}/groups'.format(base_uri)
response_data['resources'].append({'uri': resource_uri})
class GroupsDetail(APIView):
permission_classes = (ApiKeyHeaderPermission,)
# see if there is an (optional) GroupProfile
def post(self, request, group_id, format=None):
response_data = {}
base_uri = _generate_base_uri(request)
print base_uri
try:
existing_group_profile = GroupProfile.objects.get(group_id=group_id)
if existing_group_profile.group_type:
response_data['group_type'] = existing_group_profile.group_type
data = existing_group_profile.data
if data:
response_data['data'] = json.loads(data)
existing_group = Group.objects.get(id=group_id)
except ObjectDoesNotExist:
pass
response_status = status.HTTP_200_OK
return Response(response_data, status=response_status)
elif request.method == 'POST':
# update GroupProfile data
return Response({}, status.HTTP_404_NOT_FOUND)
group_type = request.DATA.get('group_type')
data = json.dumps(request.DATA.get('data')) if request.DATA.get('data') else None
if not group_type and not data:
return Response({}, status.HTTP_400_BAD_REQUEST)
profile, _ = GroupProfile.objects.get_or_create(group_id=group_id)
profile.group_type = group_type
profile.data = data
profile.save()
response_data['id'] = existing_group.id
response_data['name'] = profile.name
response_data['uri'] = _generate_base_uri(request)
return Response(response_data, status=status.HTTP_201_CREATED)
return Response({})
def get(self, request, group_id, format=None):
"""
GET retrieves an existing group from the system
"""
response_data = {}
base_uri = _generate_base_uri(request)
try:
existing_group = Group.objects.get(id=group_id)
except ObjectDoesNotExist:
return Response({}, status.HTTP_404_NOT_FOUND)
response_data['id'] = existing_group.id
response_data['uri'] = base_uri
response_data['resources'] = []
resource_uri = '{}/users'.format(base_uri)
response_data['resources'].append({'uri': resource_uri})
resource_uri = '{}/groups'.format(base_uri)
response_data['resources'].append({'uri': resource_uri})
@api_view(['GET', 'POST'])
@permission_classes((ApiKeyHeaderPermission,))
def group_users_list(request, group_id):
"""
POST creates a new group-user relationship in the system
"""
response_data = {}
group_profile = GroupProfile.objects.get(group_id=group_id)
if len(group_profile.name):
response_data['name'] = group_profile.name
else:
response_data['name'] = existing_group.name
if group_profile.group_type:
response_data['group_type'] = group_profile.group_type
data = group_profile.data
if data:
response_data['data'] = json.loads(data)
try:
existing_group = Group.objects.get(id=group_id)
except ObjectDoesNotExist:
return Response({}, status.HTTP_404_NOT_FOUND)
return Response(response_data, status=status.HTTP_200_OK)
if request.method == "GET":
users = existing_group.user_set.all()
response_data['users'] = []
for user in users:
user_data = {}
user_data['id'] = user.id
user_data['email'] = user.email
user_data['username'] = user.username
user_data['first_name'] = user.first_name
user_data['last_name'] = user.last_name
response_data['users'].append(user_data)
response_status = status.HTTP_200_OK
class GroupsUsersList(APIView):
permission_classes = (ApiKeyHeaderPermission,)
elif request.method == "POST":
user_id = request.DATA['user_id']
def post(self, request, group_id, format=None):
"""
POST creates a new group-user relationship in the system
"""
base_uri = _generate_base_uri(request)
try:
existing_group = Group.objects.get(id=group_id)
except ObjectDoesNotExist:
return Response({}, status.HTTP_404_NOT_FOUND)
user_id = request.DATA['user_id']
try:
existing_user = User.objects.get(id=user_id)
except ObjectDoesNotExist:
return Response({}, status.HTTP_404_NOT_FOUND)
try:
existing_relationship = Group.objects.get(user=existing_user)
except ObjectDoesNotExist:
existing_relationship = None
response_data = {}
if existing_relationship is None:
existing_group.user_set.add(existing_user.id)
response_data['uri'] = '{}/{}'.format(base_uri, existing_user.id)
......@@ -193,18 +187,38 @@ def group_users_list(request, group_id):
response_data['uri'] = '{}/{}'.format(base_uri, existing_user.id)
response_data['message'] = "Relationship already exists."
response_status = status.HTTP_409_CONFLICT
return Response(response_data, status=response_status)
def get(self, request, group_id, format=None):
"""
GET retrieves the list of users related to the specified group
"""
try:
existing_group = Group.objects.get(id=group_id)
except ObjectDoesNotExist:
return Response({}, status.HTTP_404_NOT_FOUND)
users = existing_group.user_set.all()
response_data = {}
response_data['users'] = []
for user in users:
user_data = {}
user_data['id'] = user.id
user_data['email'] = user.email
user_data['username'] = user.username
user_data['first_name'] = user.first_name
user_data['last_name'] = user.last_name
response_data['users'].append(user_data)
response_status = status.HTTP_200_OK
return Response(response_data, status=response_status)
return Response(response_data, status=response_status)
class GroupsUsersDetail(APIView):
permission_classes = (ApiKeyHeaderPermission,)
@api_view(['GET', 'DELETE'])
@permission_classes((ApiKeyHeaderPermission,))
def group_users_detail(request, group_id, user_id):
"""
GET retrieves an existing group-user relationship from the system
DELETE removes/inactivates/etc. an existing group-user relationship
"""
if request.method == 'GET':
def get(self, request, group_id, user_id, format=None):
"""
GET retrieves an existing group-user relationship from the system
"""
response_data = {}
base_uri = _generate_base_uri(request)
try:
......@@ -221,7 +235,12 @@ def group_users_detail(request, group_id, user_id):
else:
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
elif request.method == 'DELETE':
def delete(self, request, group_id, user_id, format=None):
"""
DELETE removes/inactivates/etc. an existing group-user relationship
"""
try:
existing_group = Group.objects.get(id=group_id)
existing_group.user_set.remove(user_id)
......@@ -231,14 +250,13 @@ def group_users_detail(request, group_id, user_id):
return Response({}, status=status.HTTP_204_NO_CONTENT)
@api_view(['POST', 'GET'])
@permission_classes((ApiKeyHeaderPermission,))
def group_groups_list(request, group_id):
"""
POST creates a new group-group relationship in the system
GET retrieves the existing group-group relationships for the specified group
"""
if request.method == 'POST':
class GroupsGroupsList(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def post(self, request, group_id, format=None):
"""
POST creates a new group-group relationship in the system
"""
response_data = {}
to_group_id = request.DATA['group_id']
relationship_type = request.DATA['relationship_type']
......@@ -266,7 +284,12 @@ def group_groups_list(request, group_id):
else:
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
elif request.method == 'GET':
def get(self, request, group_id, format=None):
"""
GET retrieves the existing group-group relationships for the specified group
"""
try:
from_group_relationship = GroupRelationship.objects.get(group__id=group_id)
except ObjectDoesNotExist:
......@@ -278,32 +301,31 @@ def group_groups_list(request, group_id):
if child_groups:
for group in child_groups:
response_data.append({
"id": group.group_id,
"relationship_type": RELATIONSHIP_TYPES['hierarchical'],
"uri": '{}/{}'.format(base_uri, group.group.id)
})
"id": group.group_id,
"relationship_type": RELATIONSHIP_TYPES['hierarchical'],
"uri": '{}/{}'.format(base_uri, group.group.id)
})
linked_groups = from_group_relationship.get_linked_group_relationships()
if linked_groups:
for group in linked_groups:
response_data.append({
"id": group.to_group_relationship_id,
"relationship_type": RELATIONSHIP_TYPES['graph'],
"uri": '{}/{}'.format(base_uri, group.to_group_relationship_id)
})
"id": group.to_group_relationship_id,
"relationship_type": RELATIONSHIP_TYPES['graph'],
"uri": '{}/{}'.format(base_uri, group.to_group_relationship_id)
})
response_status = status.HTTP_200_OK
else:
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
@api_view(['GET', 'DELETE'])
@permission_classes((ApiKeyHeaderPermission,))
def group_groups_detail(request, group_id, related_group_id):
"""
GET retrieves an existing group-group relationship from the system
DELETE removes/inactivates/etc. an existing group-group relationship
"""
if request.method == 'GET':
class GroupsGroupsDetail(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def get(self, request, group_id, related_group_id, format=None):
"""
GET retrieves an existing group-group relationship from the system
"""
response_data = {}
base_uri = _generate_base_uri(request)
response_data['uri'] = base_uri
......@@ -323,7 +345,11 @@ def group_groups_detail(request, group_id, related_group_id):
response_data['relationship_type'] = RELATIONSHIP_TYPES['graph']
response_status = status.HTTP_200_OK
return Response(response_data, response_status)
elif request.method == 'DELETE':
def delete(self, request, group_id, related_group_id, format=None):
"""
DELETE removes/inactivates/etc. an existing group-group relationship
"""
try:
from_group_relationship = GroupRelationship.objects.get(group__id=group_id)
except ObjectDoesNotExist:
......@@ -348,34 +374,19 @@ def group_groups_detail(request, group_id, related_group_id):
return Response({}, status=response_status)
@api_view(['GET', 'POST'])
@permission_classes((ApiKeyHeaderPermission,))
def group_courses_list(request, group_id):
"""
GET returns all courses that has a relationship to the group
POST creates a new group-course relationship in the system
"""
response_data = {}
try:
existing_group = Group.objects.get(id=group_id)
except ObjectDoesNotExist:
return Response({}, status.HTTP_404_NOT_FOUND)
store = modulestore()
class GroupsCoursesList(APIView):
permission_classes = (ApiKeyHeaderPermission,)
if request.method == 'GET':
members = CourseGroupRelationship.objects.filter(group=existing_group)
response_data['courses'] = []
for member in members:
course = store.get_course(member.course_id)
course_data = {
'course_id': member.course_id,
'display_name': course.display_name
}
response_data['courses'].append(course_data)
response_status = status.HTTP_200_OK
else:
def post(self, request, group_id, format=None):
"""
POST creates a new group-course relationship in the system
"""
response_data = {}
try:
existing_group = Group.objects.get(id=group_id)
except ObjectDoesNotExist:
return Response({}, status.HTTP_404_NOT_FOUND)
store = modulestore()
course_id = request.DATA['course_id']
base_uri = _generate_base_uri(request)
......@@ -398,18 +409,38 @@ def group_courses_list(request, group_id):
else:
response_data['message'] = "Relationship already exists."
response_status = status.HTTP_409_CONFLICT
return Response(response_data, status=response_status)
return Response(response_data, status=response_status)
def get(self, request, group_id, format=None):
"""
GET returns all courses that has a relationship to the group
"""
response_data = {}
try:
existing_group = Group.objects.get(id=group_id)
except ObjectDoesNotExist:
return Response({}, status.HTTP_404_NOT_FOUND)
store = modulestore()
members = CourseGroupRelationship.objects.filter(group=existing_group)
response_data['courses'] = []
for member in members:
course = store.get_course(member.course_id)
course_data = {
'course_id': member.course_id,
'display_name': course.display_name
}
response_data['courses'].append(course_data)
response_status = status.HTTP_200_OK
return Response(response_data, status=response_status)
@api_view(['GET', 'DELETE'])
@permission_classes((ApiKeyHeaderPermission,))
def group_courses_detail(request, group_id, course_id):
"""
GET retrieves an existing group-course relationship from the system
DELETE removes/inactivates/etc. an existing group-course relationship
"""
if request.method == 'GET':
class GroupsCoursesDetail(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def get(self, request, group_id, course_id, format=None):
"""
GET retrieves an existing group-course relationship from the system
"""
response_data = {}
base_uri = _generate_base_uri(request)
response_data['uri'] = base_uri
......@@ -426,7 +457,11 @@ def group_courses_detail(request, group_id, course_id):
else:
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
elif request.method == 'DELETE':
def delete(self, request, group_id, course_id, format=None):
"""
DELETE removes/inactivates/etc. an existing group-course relationship
"""
try:
existing_group = Group.objects.get(id=group_id)
existing_group.coursegrouprelationship_set.get(course_id=course_id).delete()
......
lms/djangoapps/api_manager/migrations/0003_move_name_to_profile.py 0 → 100644
# -*- coding: utf-8 -*-
import datetime
from south.db import db
from south.v2 import SchemaMigration
from django.db import models
class Migration(SchemaMigration):
def forwards(self, orm):
# Adding field 'GroupProfile.name'
db.add_column('auth_groupprofile', 'name',
self.gf('django.db.models.fields.CharField')(max_length=255, null=True, blank=True),
keep_default=False)
def backwards(self, orm):
# Deleting field 'GroupProfile.name'
db.delete_column('auth_groupprofile', 'name')
models = {
'api_manager.coursegrouprelationship': {
'Meta': {'object_name': 'CourseGroupRelationship'},
'course_id': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_index': 'True'}),
'group': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.Group']"}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'})
},
'api_manager.groupprofile': {
'Meta': {'object_name': 'GroupProfile', 'db_table': "'auth_groupprofile'"},
'data': ('django.db.models.fields.TextField', [], {'blank': 'True'}),
'group': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.Group']"}),
'group_type': ('django.db.models.fields.CharField', [], {'max_length': '32', 'null': 'True', 'db_index': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'name': ('django.db.models.fields.CharField', [], {'max_length': '255', 'null': 'True', 'blank': 'True'})
},
'api_manager.grouprelationship': {
'Meta': {'object_name': 'GroupRelationship'},
'group': ('django.db.models.fields.related.OneToOneField', [], {'to': "orm['auth.Group']", 'unique': 'True', 'primary_key': 'True'}),
'name': ('django.db.models.fields.CharField', [], {'max_length': '255'}),
'parent_group': ('django.db.models.fields.related.ForeignKey', [], {'default': '0', 'related_name': "'child_groups'", 'null': 'True', 'blank': 'True', 'to': "orm['api_manager.GroupRelationship']"}),
'record_active': ('django.db.models.fields.BooleanField', [], {'default': 'True'}),
'record_date_created': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime(2014, 4, 30, 0, 0)'}),
'record_date_modified': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'blank': 'True'})
},
'api_manager.linkedgrouprelationship': {
'Meta': {'object_name': 'LinkedGroupRelationship'},
'from_group_relationship': ('django.db.models.fields.related.ForeignKey', [], {'related_name': "'from_group_relationships'", 'to': "orm['api_manager.GroupRelationship']"}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'record_active': ('django.db.models.fields.BooleanField', [], {'default': 'True'}),
'record_date_created': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime(2014, 4, 30, 0, 0)'}),
'record_date_modified': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'blank': 'True'}),
'to_group_relationship': ('django.db.models.fields.related.ForeignKey', [], {'related_name': "'to_group_relationships'", 'to': "orm['api_manager.GroupRelationship']"})
},
'auth.group': {
'Meta': {'object_name': 'Group'},
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'name': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '80'}),
'permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'})
},
'auth.permission': {
'Meta': {'ordering': "('content_type__app_label', 'content_type__model', 'codename')", 'unique_together': "(('content_type', 'codename'),)", 'object_name': 'Permission'},
'codename': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
'content_type': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['contenttypes.ContentType']"}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'name': ('django.db.models.fields.CharField', [], {'max_length': '50'})
},
'contenttypes.contenttype': {
'Meta': {'ordering': "('name',)", 'unique_together': "(('app_label', 'model'),)", 'object_name': 'ContentType', 'db_table': "'django_content_type'"},
'app_label': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'model': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
'name': ('django.db.models.fields.CharField', [], {'max_length': '100'})
}
}
complete_apps = ['api_manager']
\ No newline at end of file
lms/djangoapps/api_manager/models.py
......@@ -105,6 +105,5 @@ class GroupProfile(models.Model):
group = models.ForeignKey(Group, db_index=True)
group_type = models.CharField(null=True, max_length=32, db_index=True)
name = models.CharField(max_length=255, null=True, blank=True)
data = models.TextField(blank=True) # JSON dictionary for generic key/value pairs
lms/djangoapps/api_manager/sessions_urls.py
""" Sessions API URI specification """
from django.conf.urls import patterns, url
urlpatterns = patterns('api_manager.sessions_views',
url(r'/*$^', 'session_list'),
url(r'^(?P<session_id>[a-z0-9]+)$', 'session_detail'),
)
from rest_framework.urlpatterns import format_suffix_patterns
from api_manager import sessions_views
urlpatterns = patterns('',
url(r'/*$^', sessions_views.SessionsList.as_view()),
url(r'^(?P<session_id>[a-z0-9]+)$', sessions_views.SessionsDetail.as_view()),
)
urlpatterns = format_suffix_patterns(urlpatterns)
lms/djangoapps/api_manager/sessions_views.py
......@@ -13,6 +13,8 @@ from django.utils.translation import ugettext as _
from rest_framework import status
from rest_framework.decorators import api_view, permission_classes
from rest_framework.response import Response
from rest_framework.views import APIView
from util.bad_request_rate_limiter import BadRequestRateLimiter
from api_manager.permissions import ApiKeyHeaderPermission
......@@ -30,81 +32,81 @@ def _generate_base_uri(request):
resource_uri = '{}://{}{}'.format(
protocol,
request.get_host(),
request.path
request.get_full_path()
)
return resource_uri
@api_view(['POST'])
@permission_classes((ApiKeyHeaderPermission,))
def session_list(request):
"""
POST creates a new system session, supported authentication modes:
1. Open edX username/password
"""
response_data = {}
# Add some rate limiting here by re-using the RateLimitMixin as a helper class
limiter = BadRequestRateLimiter()
if limiter.is_rate_limit_exceeded(request):
response_data['message'] = _('Rate limit exceeded in api login.')
return Response(response_data, status=status.HTTP_403_FORBIDDEN)
base_uri = _generate_base_uri(request)
try:
existing_user = User.objects.get(username=request.DATA['username'])
except ObjectDoesNotExist:
existing_user = None
# see if account has been locked out due to excessive login failures
if existing_user and LoginFailures.is_feature_enabled():
if LoginFailures.is_user_locked_out(existing_user):
response_status = status.HTTP_403_FORBIDDEN
response_data['message'] = _('This account has been temporarily locked due to excessive login failures. '
'Try again later.')
return Response(response_data, status=response_status)
if existing_user:
user = authenticate(username=existing_user.username, password=request.DATA['password'])
if user is not None:
# successful login, clear failed login attempts counters, if applicable
if LoginFailures.is_feature_enabled():
LoginFailures.clear_lockout_counter(user)
if user.is_active:
login(request, user)
response_data['token'] = request.session.session_key
response_data['expires'] = request.session.get_expiry_age()
user_dto = UserSerializer(user)
response_data['user'] = user_dto.data
response_data['uri'] = '{}/{}'.format(base_uri, request.session.session_key)
response_status = status.HTTP_201_CREATED
else:
class SessionsList(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def post(self, request, format=None):
"""
POST creates a new system session, supported authentication modes:
1. Open edX username/password
"""
response_data = {}
# Add some rate limiting here by re-using the RateLimitMixin as a helper class
limiter = BadRequestRateLimiter()
if limiter.is_rate_limit_exceeded(request):
response_data['message'] = _('Rate limit exceeded in api login.')
return Response(response_data, status=status.HTTP_403_FORBIDDEN)
base_uri = _generate_base_uri(request)
try:
existing_user = User.objects.get(username=request.DATA['username'])
except ObjectDoesNotExist:
existing_user = None
# see if account has been locked out due to excessive login failures
if existing_user and LoginFailures.is_feature_enabled():
if LoginFailures.is_user_locked_out(existing_user):
response_status = status.HTTP_403_FORBIDDEN
response_data['message'] = _('This account has been temporarily locked due to excessive login failures. '
'Try again later.')
return Response(response_data, status=response_status)
if existing_user:
user = authenticate(username=existing_user.username, password=request.DATA['password'])
if user is not None:
# successful login, clear failed login attempts counters, if applicable
if LoginFailures.is_feature_enabled():
LoginFailures.clear_lockout_counter(user)
if user.is_active:
login(request, user)
response_data['token'] = request.session.session_key
response_data['expires'] = request.session.get_expiry_age()
user_dto = UserSerializer(user)
response_data['user'] = user_dto.data
response_data['uri'] = '{}/{}'.format(base_uri, request.session.session_key)
response_status = status.HTTP_201_CREATED
else:
response_status = status.HTTP_401_UNAUTHORIZED
else:
limiter.tick_bad_request_counter(request)
# tick the failed login counters if the user exists in the database
if LoginFailures.is_feature_enabled():
LoginFailures.increment_lockout_counter(existing_user)
response_status = status.HTTP_401_UNAUTHORIZED
else:
limiter.tick_bad_request_counter(request)
# tick the failed login counters if the user exists in the database
if LoginFailures.is_feature_enabled():
LoginFailures.increment_lockout_counter(existing_user)
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
response_status = status.HTTP_401_UNAUTHORIZED
else:
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
class SessionsDetail(APIView):
permission_classes = (ApiKeyHeaderPermission,)
@api_view(['GET', 'DELETE'])
@permission_classes((ApiKeyHeaderPermission,))
def session_detail(request, session_id):
"""
GET retrieves an existing system session
DELETE flushes an existing system session from the system
"""
response_data = {}
base_uri = _generate_base_uri(request)
engine = import_module(settings.SESSION_ENGINE)
session = engine.SessionStore(session_id)
if request.method == 'GET':
def get(self, request, session_id, format=None):
"""
GET retrieves an existing system session
"""
response_data = {}
base_uri = _generate_base_uri(request)
engine = import_module(settings.SESSION_ENGINE)
session = engine.SessionStore(session_id)
try:
user_id = session[SESSION_KEY]
backend_path = session[BACKEND_SESSION_KEY]
......@@ -120,6 +122,14 @@ def session_detail(request, session_id):
return Response(response_data, status=status.HTTP_200_OK)
else:
return Response(response_data, status=status.HTTP_404_NOT_FOUND)
elif request.method == 'DELETE':
def delete(self, request, session_id, format=None):
"""
DELETE flushes an existing system session from the system
"""
response_data = {}
base_uri = _generate_base_uri(request)
engine = import_module(settings.SESSION_ENGINE)
session = engine.SessionStore(session_id)
session.flush()
return Response(response_data, status=status.HTTP_204_NO_CONTENT)
lms/djangoapps/api_manager/system_views.py
......@@ -3,6 +3,7 @@
from rest_framework import status
from rest_framework.decorators import api_view, permission_classes
from rest_framework.response import Response
from rest_framework.views import APIView
from api_manager.permissions import ApiKeyHeaderPermission
......@@ -17,38 +18,39 @@ def _generate_base_uri(request):
resource_uri = '{}://{}{}'.format(
protocol,
request.get_host(),
request.path
request.get_full_path()
)
return resource_uri
@api_view(['GET'])
@permission_classes((ApiKeyHeaderPermission,))
def system_detail(request):
"""Returns top-level descriptive information about the Open edX API"""
base_uri = _generate_base_uri(request)
response_data = {}
response_data['name'] = "Open edX System API"
response_data['description'] = "System interface for managing groups, users, and sessions."
response_data['documentation'] = "http://docs.openedxapi.apiary.io/#get-%2Fapi%2Fsystem"
response_data['uri'] = base_uri
return Response(response_data, status=status.HTTP_200_OK)
@api_view(['GET'])
@permission_classes((ApiKeyHeaderPermission,))
def api_detail(request):
"""Returns top-level descriptive information about the Open edX API"""
base_uri = _generate_base_uri(request)
response_data = {}
response_data['name'] = "Open edX API"
response_data['description'] = "Machine interface for interactions with Open edX."
response_data['documentation'] = "http://docs.openedxapi.apiary.io"
response_data['uri'] = base_uri
response_data['resources'] = []
response_data['resources'].append({'uri': base_uri + 'courses'})
response_data['resources'].append({'uri': base_uri + 'groups'})
response_data['resources'].append({'uri': base_uri + 'sessions'})
response_data['resources'].append({'uri': base_uri + 'system'})
response_data['resources'].append({'uri': base_uri + 'users'})
return Response(response_data, status=status.HTTP_200_OK)
class SystemDetail(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def get(self, request, format=None):
"""Returns top-level descriptive information about the Open edX API"""
base_uri = _generate_base_uri(request)
response_data = {}
response_data['name'] = "Open edX System API"
response_data['description'] = "System interface for managing groups, users, and sessions."
response_data['documentation'] = "http://docs.openedxapi.apiary.io/#get-%2Fapi%2Fsystem"
response_data['uri'] = base_uri
return Response(response_data, status=status.HTTP_200_OK)
class ApiDetail(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def get(self, request, format=None):
"""Returns top-level descriptive information about the Open edX API"""
base_uri = _generate_base_uri(request)
response_data = {}
response_data['name'] = "Open edX API"
response_data['description'] = "Machine interface for interactions with Open edX."
response_data['documentation'] = "http://docs.openedxapi.apiary.io"
response_data['uri'] = base_uri
response_data['resources'] = []
response_data['resources'].append({'uri': base_uri + 'courses'})
response_data['resources'].append({'uri': base_uri + 'groups'})
response_data['resources'].append({'uri': base_uri + 'sessions'})
response_data['resources'].append({'uri': base_uri + 'system'})
response_data['resources'].append({'uri': base_uri + 'users'})
return Response(response_data, status=status.HTTP_200_OK)
lms/djangoapps/api_manager/tests/test_courses_views.py
......@@ -136,7 +136,13 @@ class CoursesApiTests(TestCase):
response = self.client.delete(uri, headers=headers)
return response
def test_course_list_get(self):
def _find_item_by_class(self, items, class_name):
for item in items:
if item['class'] == class_name:
return item
return None
def test_courses_list_get(self):
test_uri = self.base_courses_uri
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
......@@ -152,7 +158,7 @@ class CoursesApiTests(TestCase):
matched_course = True
self.assertTrue(matched_course)
def test_course_detail_get(self):
def test_courses_detail_get(self):
test_uri = self.base_courses_uri + '/' + self.test_course_id
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
......@@ -163,16 +169,28 @@ class CoursesApiTests(TestCase):
self.assertEqual(response.data['org'], self.test_course_org)
confirm_uri = self.test_server_prefix + test_uri
self.assertEqual(response.data['uri'], confirm_uri)
def test_courses_detail_get_with_submodules(self):
test_uri = self.base_courses_uri + '/' + self.test_course_id + '?depth=100'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
self.assertEqual(response.data['id'], self.test_course_id)
self.assertEqual(response.data['name'], self.test_course_name)
self.assertEqual(response.data['number'], self.test_course_number)
self.assertEqual(response.data['org'], self.test_course_org)
confirm_uri = self.test_server_prefix + test_uri
self.assertEqual(response.data['uri'], confirm_uri)
self.assertGreater(len(response.data['modules']), 0)
def test_course_detail_get_notfound(self):
test_uri = self.base_courses_uri + '/' + 'p29038cvp9hjwefion'
def test_courses_detail_get_notfound(self):
test_uri = self.base_courses_uri + '/' + self.test_bogus_course_id
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_course_tree_get(self):
def test_courses_tree_get(self):
# query the course tree to quickly get naviation information
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/tree/2'
test_uri = self.base_courses_uri + '/' + self.test_course_id + '?depth=2'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
......@@ -190,9 +208,9 @@ class CoursesApiTests(TestCase):
self.assertEqual(sequence['name'], 'Video_Sequence')
self.assertNotIn('modules', sequence)
def test_course_tree_get_root(self):
def test_courses_tree_get_root(self):
# query the course tree to quickly get naviation information
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/tree/0'
test_uri = self.base_courses_uri + '/' + self.test_course_id + '?depth=0'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
......@@ -250,6 +268,16 @@ class CoursesApiTests(TestCase):
self.assertEqual(response.data['uri'], confirm_uri)
self.assertGreater(len(response.data['modules']), 0)
def test_modules_detail_get_course(self):
test_uri = self.base_modules_uri + '/' + self.test_course_id
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
self.assertEqual(response.data['id'], self.test_course_id)
confirm_uri = self.test_server_prefix + self.base_courses_uri + '/' + self.test_course_id
self.assertEqual(response.data['uri'], confirm_uri)
self.assertGreater(len(response.data['modules']), 0)
def test_modules_detail_get_notfound(self):
test_uri = self.base_modules_uri + '/' + '2p38fp2hjfp9283'
response = self.do_get(test_uri)
......@@ -273,7 +301,7 @@ class CoursesApiTests(TestCase):
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_course_groups_list_post(self):
def test_courses_groups_list_post(self):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
group_id = response.data['id']
......@@ -288,7 +316,7 @@ class CoursesApiTests(TestCase):
self.assertEqual(response.data['course_id'], str(self.test_course_id))
self.assertEqual(response.data['group_id'], str(group_id))
def test_course_groups_list_post_duplicate(self):
def test_courses_groups_list_post_duplicate(self):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
group_id = response.data['id']
......@@ -299,13 +327,13 @@ class CoursesApiTests(TestCase):
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 409)
def test_group_courses_list_post_invalid_resources(self):
test_uri = self.base_courses_uri + '/1239878976/groups'
def test_courses_groups_list_post_invalid_resources(self):
test_uri = self.base_courses_uri + '/1239/87/8976/groups'
data = {'group_id': "98723896"}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 404)
def test_course_groups_detail_get(self):
def test_courses_groups_detail_get(self):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
group_id = response.data['id']
......@@ -319,7 +347,18 @@ class CoursesApiTests(TestCase):
self.assertEqual(response.data['course_id'], self.test_course_id)
self.assertEqual(response.data['group_id'], str(group_id))
def test_course_groups_detail_delete(self):
def test_courses_groups_detail_get_invalid_resources(self):
course_id = 'asd/fas/vcsadfaf'
group_id = '12343'
test_uri = '{}/{}/groups/{}'.format(self.base_courses_uri, course_id, group_id)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
confirm_uri = self.test_server_prefix + test_uri
self.assertEqual(response.data['uri'], confirm_uri)
self.assertEqual(response.data['course_id'], course_id)
self.assertEqual(response.data['group_id'], group_id)
def test_courses_groups_detail_delete(self):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
test_uri = '{}/{}/groups'.format(self.base_courses_uri, self.test_course_id)
......@@ -333,17 +372,17 @@ class CoursesApiTests(TestCase):
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_course_groups_detail_delete_invalid_course(self):
def test_courses_groups_detail_delete_invalid_course(self):
test_uri = '{}/{}/groups/123124'.format(self.base_courses_uri, self.test_bogus_course_id)
response = self.do_delete(test_uri)
self.assertEqual(response.status_code, 204)
def test_course_groups_detail_delete_invalid_group(self):
def test_courses_groups_detail_delete_invalid_group(self):
test_uri = '{}/{}/groups/123124'.format(self.base_courses_uri, self.test_course_id)
response = self.do_delete(test_uri)
self.assertEqual(response.status_code, 204)
def test_course_groups_detail_get_undefined(self):
def test_courses_groups_detail_get_undefined(self):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
group_id = response.data['id']
......@@ -351,7 +390,7 @@ class CoursesApiTests(TestCase):
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_get_course_overview_unparsed(self):
def test_courses_overview_get_unparsed(self):
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/overview'
response = self.do_get(test_uri)
......@@ -359,13 +398,7 @@ class CoursesApiTests(TestCase):
self.assertGreater(len(response.data), 0)
self.assertEqual(response.data['overview_html'], self.overview.data)
def _find_item_by_class(self, items, class_name):
for item in items:
if item['class'] == class_name:
return item
return None
def test_get_course_overview_parsed(self):
def test_courses_overview_get_parsed(self):
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/overview?parse=true'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
......@@ -393,8 +426,29 @@ class CoursesApiTests(TestCase):
self.assertGreater(len(prerequisites['body']), 0)
faq = self._find_item_by_class(sections, 'faq')
self.assertGreater(len(faq['body']), 0)
invalid_tab = self._find_item_by_class(sections, 'invalid_tab')
self.assertFalse(invalid_tab)
def test_courses_overview_get_invalid_course(self):
#try a bogus course_id to test failure case
test_uri = '{}/{}/overview'.format(self.base_courses_uri, self.test_bogus_course_id)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_get_course_updates(self):
def test_courses_overview_get_invalid_content(self):
#try a bogus course_id to test failure case
test_course = CourseFactory.create()
test_uri = '{}/{}/overview'.format(self.base_courses_uri, test_course.id)
test_updates = ItemFactory.create(
category="about",
parent_location=test_course.location,
data='',
display_name="overview"
)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_courses_updates_get(self):
# first try raw without any parsing
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/updates'
response = self.do_get(test_uri)
......@@ -419,8 +473,28 @@ class CoursesApiTests(TestCase):
self.assertEqual(postings[3]['date'], 'April 15, 2014')
self.assertEqual(postings[3]['content'], '<p>A perfectly</p><p>formatted piece</p><p>of HTML</p>')
def test_static_tab_list(self):
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/static_tabs'
def test_courses_updates_get_invalid_course(self):
#try a bogus course_id to test failure case
test_uri = '{}/{}/updates'.format(self.base_courses_uri, self.test_bogus_course_id)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_courses_updates_get_invalid_content(self):
#try a bogus course_id to test failure case
test_course = CourseFactory.create()
test_course_data = '<html>{}</html>'.format(str(uuid.uuid4()))
test_updates = ItemFactory.create(
category="course_info",
parent_location=test_course.location,
data='',
display_name="updates"
)
test_uri = '{}/{}/updates'.format(self.base_courses_uri, test_course.id)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_static_tab_list_get(self):
test_uri = '{}/{}/static_tabs'.format(self.base_courses_uri, self.test_course_id)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
......@@ -447,12 +521,13 @@ class CoursesApiTests(TestCase):
self.assertEqual(tabs[1]['id'], u'readings')
self.assertEqual(tabs[1]['content'], self.static_tab2.data)
def test_static_tab_list_get_invalid_course(self):
#try a bogus course_id to test failure case
test_uri = self.base_courses_uri + '/' + self.test_bogus_course_id + '/static_tabs'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_static_tab_detail(self):
def test_static_tab_detail_get(self):
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/static_tabs/syllabus'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
......@@ -471,17 +546,19 @@ class CoursesApiTests(TestCase):
self.assertEqual(tab['id'], u'readings')
self.assertEqual(tab['content'], self.static_tab2.data)
def test_static_tab_detail_get_invalid_course(self):
# try a bogus courseId
test_uri = self.base_courses_uri + '/' + self.test_bogus_course_id + '/static_tabs/syllabus'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_static_tab_detail_get_invalid_item(self):
# try a not found item
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/static_tabs/bogus'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_course_enrollments(self):
def test_courses_users_list_get(self):
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/users'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
......@@ -492,32 +569,40 @@ class CoursesApiTests(TestCase):
self.assertEqual(len(enrollments), 0)
self.assertNotIn('pending_enrollments', response.data)
def test_courses_users_list_invalid_course(self):
test_uri = self.base_courses_uri + '/' + self.test_bogus_course_id + '/users'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_courses_users_list_post_nonexisting_user_deny(self):
# enroll a non-existing student
# first, don't allow non-existing
post_data = {}
post_data['email'] = 'test+pending@tester.com'
post_data['allow_pending'] = False
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/users'
post_data = {
'email': 'test+pending@tester.com',
'allow_pending': False,
}
response = self.do_post(test_uri, post_data)
self.assertEqual(response.status_code, 400)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
def test_courses_users_list_post_nonexisting_user_allow(self):
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/users'
post_data = {}
post_data['email'] = 'test+pending@tester.com'
post_data['allow_pending'] = True
response = self.do_post(test_uri, post_data)
self.assertEqual(response.status_code, 201)
# re-run query
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
self.assertEqual(len(response.data['enrollments']), 0)
# assert that we just have a single pending enrollment
enrollments = response.data['enrollments']
self.assertEqual(len(enrollments), 0)
self.assertIn('pending_enrollments', response.data)
pending = response.data['pending_enrollments']
self.assertEqual(len(pending), 1)
self.assertEqual(pending[0], 'test+pending@tester.com')
def test_courses_users_list_post_existing_user(self):
# create a new user (note, this calls into the /users/ subsystem)
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/users'
test_user_uri = '/api/users'
local_username = "some_test_user" + str(randint(11, 99))
local_email = "test+notpending@tester.com"
......@@ -533,21 +618,131 @@ class CoursesApiTests(TestCase):
self.assertGreater(response.data['id'], 0)
created_user_id = response.data['id']
# now register this user
# now enroll this user in the course
post_data = {}
post_data['user_id'] = created_user_id
response = self.do_post(test_uri, post_data)
self.assertEqual(response.status_code, 201)
# now re-query, we should see it listed now in the list of enrollments
# re-run query
def test_courses_users_list_post_invalid_course(self):
test_uri = self.base_courses_uri + '/' + self.test_bogus_course_id + '/users'
post_data = {}
post_data['email'] = 'test+pending@tester.com'
post_data['allow_pending'] = True
response = self.do_post(test_uri, post_data)
self.assertEqual(response.status_code, 404)
def test_courses_users_list_post_invalid_user(self):
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/users'
post_data = {}
post_data['user_id'] = '123123124'
post_data['allow_pending'] = True
response = self.do_post(test_uri, post_data)
self.assertEqual(response.status_code, 404)
def test_courses_users_list_post_invalid_payload(self):
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/users'
post_data = {}
response = self.do_post(test_uri, post_data)
self.assertEqual(response.status_code, 400)
def test_courses_users_list_get(self):
# create a new user (note, this calls into the /users/ subsystem)
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/users'
test_user_uri = '/api/users'
local_username = "some_test_user" + str(randint(11, 99))
local_email = "test+notpending@tester.com"
data = {
'email': local_email,
'username': local_username,
'password': 'fooabr',
'first_name': 'Joe',
'last_name': 'Brown'
}
response = self.do_post(test_user_uri, data)
self.assertEqual(response.status_code, 201)
self.assertGreater(response.data['id'], 0)
created_user_id = response.data['id']
post_data = {}
post_data['user_id'] = created_user_id
response = self.do_post(test_uri, post_data)
self.assertEqual(response.status_code, 201)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
def test_courses_users_detail_get(self):
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/users'
test_user_uri = '/api/users'
local_username = "some_test_user" + str(randint(11, 99))
local_email = "test+notpending@tester.com"
data = {
'email': local_email,
'username': local_username,
'password': 'fooabr',
'first_name': 'Joe',
'last_name': 'Brown'
}
response = self.do_post(test_user_uri, data)
self.assertEqual(response.status_code, 201)
self.assertGreater(response.data['id'], 0)
created_user_id = response.data['id']
# now enroll this user in the course
post_data = {}
post_data['user_id'] = created_user_id
response = self.do_post(test_uri, post_data)
self.assertEqual(response.status_code, 201)
confirm_uri = '{}/{}'.format(test_uri, created_user_id)
response = self.do_get(confirm_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
# assert that we just have a single pending enrollment
enrollments = response.data['enrollments']
self.assertEqual(len(enrollments), 1)
self.assertEqual(enrollments[0]['id'], created_user_id)
self.assertEqual(enrollments[0]['email'], local_email)
self.assertEqual(enrollments[0]['username'], local_username)
def test_courses_users_detail_get_invalid_course(self):
test_uri = self.base_courses_uri + '/' + self.test_bogus_course_id + '/users/213432'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
self.assertGreater(len(response.data), 0)
def test_courses_users_detail_get_invalid_user(self):
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/users/213432'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
self.assertGreater(len(response.data), 0)
def test_courses_users_detail_delete(self):
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/users'
test_user_uri = '/api/users'
local_username = "some_test_user" + str(randint(11, 99))
local_email = "test+notpending@tester.com"
data = {
'email': local_email,
'username': local_username,
'password': 'fooabr',
'first_name': 'Joe',
'last_name': 'Brown'
}
response = self.do_post(test_user_uri, data)
self.assertEqual(response.status_code, 201)
self.assertGreater(response.data['id'], 0)
created_user_id = response.data['id']
# now enroll this user in the course
post_data = {}
post_data['user_id'] = created_user_id
response = self.do_post(test_uri, post_data)
self.assertEqual(response.status_code, 201)
confirm_uri = '{}/{}'.format(test_uri, created_user_id)
response = self.do_get(confirm_uri)
self.assertEqual(response.status_code, 200)
response = self.do_delete(confirm_uri)
self.assertEqual(response.status_code, 204)
def test_courses_users_detail_delete_invalid_course(self):
test_uri = self.base_courses_uri + '/' + self.test_bogus_course_id + '/users/213432'
response = self.do_delete(test_uri)
self.assertEqual(response.status_code, 404)
def test_courses_users_detail_delete_invalid_user(self):
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/users/213432'
response = self.do_delete(test_uri)
self.assertEqual(response.status_code, 204)
lms/djangoapps/api_manager/tests/test_groups_views.py
......@@ -83,30 +83,6 @@ class GroupsApiTests(TestCase):
self.assertEqual(response.data['uri'], confirm_uri)
self.assertGreater(len(response.data['name']), 0)
# @unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
# def test_group_list_post_duplicate(self):
# data = {'name': self.test_group_name}
# response = self.do_post(self.base_groups_uri, data)
# self.assertEqual(response.status_code, 201)
# self.assertGreater(response.data['id'], 0)
# confirm_uri = self.test_server_prefix + self.base_groups_uri + '/' + str(response.data['id'])
# self.assertEqual(response.data['uri'], confirm_uri)
# response = self.do_post(self.base_groups_uri, data)
# self.assertEqual(response.status_code, 409)
def test_group_detail_get(self):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
self.assertGreater(response.data['id'], 0)
group_id = response.data['id']
test_uri = self.base_groups_uri + '/' + str(group_id)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data['id'], group_id)
confirm_uri = self.test_server_prefix + test_uri
self.assertEqual(response.data['uri'], confirm_uri)
self.assertEqual(response.data['name'], self.test_group_name)
def test_group_list_get_with_profile(self):
data = {
'name': self.test_group_name,
......@@ -119,10 +95,10 @@ class GroupsApiTests(TestCase):
self.assertGreater(response.data['id'], 0)
group_id = response.data['id']
# query for list of groups, but don't put the type filter (bad)
# query for list of groups, but don't put the type filter
test_uri = self.base_groups_uri
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 400)
self.assertEqual(response.status_code, 200)
# try again with filter
test_uri = self.base_groups_uri + '?type=series'
......@@ -132,6 +108,7 @@ class GroupsApiTests(TestCase):
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]['group_id'], group_id)
self.assertEqual(response.data[0]['group_type'], 'series')
self.assertEqual(response.data[0]['name'], self.test_group_name)
self.assertEqual(response.data[0]['data']['display_name'], 'My first series')
# query the group detail
......@@ -159,7 +136,7 @@ class GroupsApiTests(TestCase):
}
}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.status_code, 201)
# requery the filter
test_uri = self.base_groups_uri + '?type=series'
......@@ -173,13 +150,84 @@ class GroupsApiTests(TestCase):
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]['group_id'], group_id)
self.assertEqual(response.data[0]['group_type'], 'seriesX')
self.assertEqual(response.data[0]['name'], self.test_group_name)
self.assertEqual(response.data[0]['data']['display_name'], 'My updated series')
def test_group_list_get_uses_base_group_name(self):
data = {'name': ''}
response = self.do_post(self.base_groups_uri, data)
self.assertEqual(response.status_code, 201)
group_id = response.data['id']
response = self.do_get(self.base_groups_uri)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data[0]['name'], '{:04d}: '.format(group_id))
def test_group_detail_get(self):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
self.assertEqual(response.status_code, 201)
self.assertGreater(response.data['id'], 0)
group_id = response.data['id']
test_uri = self.base_groups_uri + '/' + str(group_id)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data['id'], group_id)
confirm_uri = self.test_server_prefix + test_uri
self.assertEqual(response.data['uri'], confirm_uri)
self.assertEqual(response.data['name'], self.test_group_name)
def test_group_detail_get_uses_base_group_name(self):
data = {'name': ''}
response = self.do_post(self.base_groups_uri, data)
self.assertEqual(response.status_code, 201)
self.assertGreater(response.data['id'], 0)
group_id = response.data['id']
test_uri = self.base_groups_uri + '/' + str(group_id)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data['id'], group_id)
confirm_uri = self.test_server_prefix + test_uri
self.assertEqual(response.data['uri'], confirm_uri)
self.assertEqual(response.data['name'], '{:04d}: '.format(group_id))
def test_group_detail_get_undefined(self):
test_uri = self.base_groups_uri + '/123456789'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_group_detail_post(self):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
group_id = response.data['id']
test_uri = response.data['uri']
self.assertEqual(response.status_code, 201)
group_type = 'seriesX'
data = {
'name': self.test_group_name,
'group_type': group_type,
'data': {
'display_name': 'My updated series'
}
}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 201)
self.assertEqual(response.data['id'], group_id)
self.assertEqual(response.data['name'], self.test_group_name)
self.assertEqual(response.data['uri'], test_uri)
def test_group_detail_post_invalid_group(self):
test_uri = '{}/23209232'.format(self.base_groups_uri)
group_type = 'seriesX'
data = {
'name': self.test_group_name,
'group_type': group_type,
'data': {
'display_name': 'My updated series'
}
}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 404)
def test_group_users_list_post(self):
local_username = self.test_username + str(randint(11, 99))
data = {
......@@ -205,17 +253,6 @@ class GroupsApiTests(TestCase):
self.assertEqual(response.data['group_id'], str(group_id))
self.assertEqual(response.data['user_id'], str(user_id))
# check to see if the user is listed after we associate it with the group
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
users = response.data['users']
self.assertEqual(len(users), 1)
self.assertEqual(users[0]['id'], user_id)
self.assertEqual(users[0]['username'], local_username)
self.assertEqual(users[0]['email'], self.test_email)
self.assertEqual(users[0]['first_name'], 'Joe')
self.assertEqual(users[0]['last_name'], 'Smith')
def test_group_users_list_post_duplicate(self):
local_username = self.test_username + str(randint(11, 99))
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password}
......@@ -232,13 +269,57 @@ class GroupsApiTests(TestCase):
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 409)
def test_group_users_list_post_invalid_resources(self):
def test_group_users_list_post_invalid_group(self):
test_uri = self.base_groups_uri + '/1239878976'
test_uri = test_uri + '/users'
data = {'user_id': "98723896"}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 404)
def test_group_users_list_post_invalid_user(self):
data = {'name': 'Alpha Group'}
response = self.do_post(self.base_groups_uri, data)
test_uri = '{}/{}/users'.format(self.base_groups_uri, str(response.data['id']))
data = {'user_id': "98723896"}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 404)
def test_group_users_list_get(self):
local_username = self.test_username + str(randint(11, 99))
data = {
'email': self.test_email,
'username': local_username,
'password': self.test_password,
'first_name': 'Joe',
'last_name': 'Smith'
}
response = self.do_post(self.base_users_uri, data)
user_id = response.data['id']
data = {'name': 'Alpha Group'}
response = self.do_post(self.base_groups_uri, data)
group_id = response.data['id']
test_uri = self.base_groups_uri + '/' + str(group_id)
response = self.do_get(test_uri)
test_uri = test_uri + '/users'
data = {'user_id': user_id}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 201)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
users = response.data['users']
self.assertEqual(len(users), 1)
self.assertEqual(users[0]['id'], user_id)
self.assertEqual(users[0]['username'], local_username)
self.assertEqual(users[0]['email'], self.test_email)
self.assertEqual(users[0]['first_name'], 'Joe')
self.assertEqual(users[0]['last_name'], 'Smith')
def test_group_users_list_get_invalid_group(self):
test_uri = self.base_groups_uri + '/1231241/users'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_group_users_detail_get(self):
local_username = self.test_username + str(randint(11, 99))
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password}
......@@ -592,13 +673,6 @@ class GroupsApiTests(TestCase):
self.assertEqual(response.data['group_id'], str(group_id))
self.assertEqual(response.data['course_id'], self.test_course_id)
# then re-query to check membership
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['courses']), 1)
self.assertEqual(response.data['courses'][0]['course_id'], self.test_course_id)
self.assertEqual(response.data['courses'][0]['display_name'], self.course.display_name)
def test_group_courses_list_post_duplicate(self):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
......@@ -617,6 +691,30 @@ class GroupsApiTests(TestCase):
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 404)
def test_group_courses_list_get(self):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
self.assertEqual(response.status_code, 201)
group_id = response.data['id']
test_uri = response.data['uri'] + '/courses'
data = {'course_id': self.test_course_id}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 201)
confirm_uri = test_uri + '/' + self.course.id
self.assertEqual(response.data['uri'], confirm_uri)
self.assertEqual(response.data['group_id'], str(group_id))
self.assertEqual(response.data['course_id'], self.test_course_id)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['courses']), 1)
self.assertEqual(response.data['courses'][0]['course_id'], self.test_course_id)
self.assertEqual(response.data['courses'][0]['display_name'], self.course.display_name)
def test_group_courses_list_get_invalid_group(self):
test_uri = self.base_groups_uri + '/1231241/courses'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_group_courses_detail_get(self):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
......
lms/djangoapps/api_manager/tests/test_permissions.py
......@@ -30,7 +30,6 @@ class PermissionsTestsDebug(TestCase):
response = self.client.post(uri, headers=headers, data=data)
return response
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_has_permission_debug_enabled(self):
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
......@@ -57,7 +56,6 @@ class PermissionsTestsApiKey(TestCase):
response = self.client.post(uri, headers=headers, data=data)
return response
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_has_permission_valid_api_key(self):
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
......@@ -84,7 +82,6 @@ class PermissionsTestDeniedMissingServerKey(TestCase):
response = self.client.post(uri, headers=headers, data=data)
return response
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_has_permission_missing_server_key(self):
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
......@@ -110,7 +107,6 @@ class PermissionsTestDeniedMissingClientKey(TestCase):
response = self.client.post(uri, headers=headers, data=data)
return response
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_has_permission_invalid_client_key(self):
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
......@@ -137,7 +133,6 @@ class PermissionsTestDeniedInvalidClientKey(TestCase):
response = self.client.post(uri, headers=headers, data=data)
return response
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_has_permission_invalid_client_key(self):
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
......
lms/djangoapps/api_manager/tests/test_sessions_views.py
......@@ -68,7 +68,6 @@ class SessionsApiTests(TestCase):
response = self.client.delete(uri, headers=headers)
return response
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_session_list_post_valid(self):
local_username = self.test_username + str(randint(11, 99))
local_username = local_username[3:-1] # username is a 32-character field
......@@ -86,7 +85,6 @@ class SessionsApiTests(TestCase):
self.assertEqual(str(response.data['user']['username']), local_username)
self.assertEqual(response.data['user']['id'], user_id)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_session_list_post_invalid(self):
local_username = self.test_username + str(randint(11, 99))
local_username = local_username[3:-1] # username is a 32-character field
......@@ -97,7 +95,6 @@ class SessionsApiTests(TestCase):
response = self.do_post(self.base_sessions_uri, data)
self.assertEqual(response.status_code, 401)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_session_list_post_valid_inactive(self):
local_username = self.test_username + str(randint(11, 99))
local_username = local_username[3:-1] # username is a 32-character field
......@@ -108,15 +105,13 @@ class SessionsApiTests(TestCase):
user.save()
data = {'username': local_username, 'password': self.test_password}
response = self.do_post(self.base_sessions_uri, data)
self.assertEqual(response.status_code, 403)
self.assertEqual(response.status_code, 401)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_session_list_post_invalid_notfound(self):
data = {'username': 'user_12321452334', 'password': self.test_password}
response = self.do_post(self.base_sessions_uri, data)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_session_detail_get(self):
local_username = self.test_username + str(randint(11, 99))
local_username = local_username[3:-1] # username is a 32-character field
......@@ -134,13 +129,11 @@ class SessionsApiTests(TestCase):
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_session_detail_get_undefined(self):
test_uri = self.base_sessions_uri + "/123456789"
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_session_detail_delete(self):
local_username = self.test_username + str(randint(11, 99))
local_username = local_username[3:-1] # username is a 32-character field
......
lms/djangoapps/api_manager/tests/test_users_views.py
......@@ -73,7 +73,6 @@ class UsersApiTests(TestCase):
response = self.client.delete(uri, headers=headers)
return response
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_list_post(self):
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
......@@ -88,7 +87,6 @@ class UsersApiTests(TestCase):
self.assertEqual(response.data['first_name'], self.test_first_name)
self.assertEqual(response.data['last_name'], self.test_last_name)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_list_post_duplicate(self):
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
......@@ -99,7 +97,6 @@ class UsersApiTests(TestCase):
self.assertGreater(response.data['message'], 0)
self.assertEqual(response.data['field_conflict'], 'username')
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_detail_get(self):
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
......@@ -117,7 +114,6 @@ class UsersApiTests(TestCase):
self.assertEqual(response.data['last_name'], self.test_last_name)
self.assertEqual(len(response.data['resources']), 2)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_detail_delete(self):
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
......@@ -131,13 +127,11 @@ class UsersApiTests(TestCase):
response = self.do_delete(test_uri) # User no longer exists, should get a 204 all the same
self.assertEqual(response.status_code, 204)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_detail_get_undefined(self):
test_uri = '/api/users/123456789'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_groups_list_post(self):
test_uri = '/api/groups'
data = {'name': 'Alpha Group'}
......@@ -160,7 +154,6 @@ class UsersApiTests(TestCase):
self.assertEqual(response.data['group_id'], str(group_id))
self.assertEqual(response.data['user_id'], str(user_id))
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_groups_list_post_duplicate(self):
test_uri = '/api/groups'
data = {'name': 'Alpha Group'}
......@@ -179,7 +172,6 @@ class UsersApiTests(TestCase):
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 409)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_groups_list_post_invalid_user(self):
test_uri = '/api/groups'
data = {'name': 'Alpha Group'}
......@@ -190,7 +182,34 @@ class UsersApiTests(TestCase):
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_groups_list_get(self):
test_uri = '/api/groups'
group_name = 'Alpha Group'
data = {'name': group_name}
response = self.do_post(test_uri, data)
group_id = response.data['id']
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password, 'first_name': self.test_first_name, 'last_name': self.test_last_name}
response = self.do_post(test_uri, data)
user_id = response.data['id']
test_uri = test_uri + '/' + str(response.data['id'])
response = self.do_get(test_uri)
test_uri = test_uri + '/groups'
data = {'group_id': group_id}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 201)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data['groups']), 0)
self.assertEqual(response.data['groups'][0]['id'], group_id)
self.assertEqual(response.data['groups'][0]['name'], str(group_name))
def test_user_groups_list_get_invalid_user(self):
test_uri = '/api/users/123124/groups'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_user_groups_detail_get(self):
test_uri = '/api/groups'
data = {'name': 'Alpha Group'}
......@@ -213,7 +232,6 @@ class UsersApiTests(TestCase):
self.assertEqual(response.data['group_id'], group_id)
self.assertEqual(response.data['user_id'], user_id)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_groups_detail_delete(self):
test_uri = '/api/groups'
data = {'name': 'Alpha Group'}
......@@ -236,7 +254,11 @@ class UsersApiTests(TestCase):
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_groups_detail_get_invalid_user(self):
test_uri = '/api/users/123124/groups/12321'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_user_groups_detail_get_undefined(self):
test_uri = '/api/groups'
data = {'name': 'Alpha Group'}
......@@ -251,7 +273,6 @@ class UsersApiTests(TestCase):
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_courses_list_post(self):
course = CourseFactory.create()
test_uri = '/api/users'
......@@ -268,7 +289,6 @@ class UsersApiTests(TestCase):
self.assertEqual(response.data['id'], course.id)
self.assertTrue(response.data['is_active'])
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_courses_list_post_undefined_user(self):
course = CourseFactory.create()
test_uri = '/api/users'
......@@ -278,7 +298,6 @@ class UsersApiTests(TestCase):
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_courses_list_post_undefined_course(self):
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
......@@ -290,7 +309,6 @@ class UsersApiTests(TestCase):
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_courses_list_get(self):
course = CourseFactory.create(display_name="TEST COURSE")
test_uri = '/api/users'
......@@ -310,13 +328,11 @@ class UsersApiTests(TestCase):
self.assertTrue(response.data[0]['is_active'])
self.assertEqual(response.data[0]['name'], course.display_name)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_courses_list_get_undefined_user(self):
test_uri = '/api/users/2134234/courses'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_courses_detail_post_position_course_as_descriptor(self):
course = CourseFactory.create()
test_data = '<html>{}</html>'.format(str(uuid.uuid4()))
......@@ -358,7 +374,6 @@ class UsersApiTests(TestCase):
response = self.do_post(test_uri, data=position_data)
self.assertEqual(response.data['position'], chapter3.id)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_courses_detail_post_position_invalid_user(self):
course = CourseFactory.create()
test_data = '<html>{}</html>'.format(str(uuid.uuid4()))
......@@ -381,7 +396,6 @@ class UsersApiTests(TestCase):
response = self.do_post(test_uri, data=position_data)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_courses_detail_post_position_course_as_module(self):
course = CourseFactory.create()
test_data = '<html>{}</html>'.format(str(uuid.uuid4()))
......@@ -411,7 +425,6 @@ class UsersApiTests(TestCase):
response = self.do_post(test_uri, data=position_data)
self.assertEqual(response.data['position'], chapter1.id)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_courses_detail_get(self):
course = CourseFactory.create()
test_data = '<html>{}</html>'.format(str(uuid.uuid4()))
......@@ -447,13 +460,11 @@ class UsersApiTests(TestCase):
response = self.do_post(confirm_uri, data=position_data)
self.assertEqual(response.data['position'], chapter1.id)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_courses_detail_get_undefined_user(self):
test_uri = '/api/users/2134234/courses/a8df7asvd98'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_courses_detail_get_undefined_enrollment(self):
course = CourseFactory.create()
test_uri = '/api/users'
......@@ -465,7 +476,6 @@ class UsersApiTests(TestCase):
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_courses_detail_delete(self):
course = CourseFactory.create()
test_uri = '/api/users'
......@@ -493,7 +503,6 @@ class UsersApiTests(TestCase):
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_courses_detail_delete_undefined_user(self):
course = CourseFactory.create()
user_id = '2134234'
......@@ -501,7 +510,6 @@ class UsersApiTests(TestCase):
response = self.do_delete(test_uri)
self.assertEqual(response.status_code, 204)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_user_courses_detail_delete_undefined_course(self):
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
......
lms/djangoapps/api_manager/tests/test_views.py
......@@ -46,7 +46,6 @@ class SystemApiTests(TestCase):
response = self.client.get(uri, headers=headers)
return response
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_system_detail_get(self):
""" Ensure the system returns base data about the system """
test_uri = self.test_server_prefix + '/system'
......@@ -62,7 +61,6 @@ class SystemApiTests(TestCase):
self.assertIsNotNone(response.data['description'])
self.assertGreater(len(response.data['description']), 0)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_system_detail_api_get(self):
""" Ensure the system returns base data about the API """
test_uri = self.test_server_prefix
......
lms/djangoapps/api_manager/urls.py
......@@ -10,9 +10,11 @@
from django.conf.urls import include, patterns, url
urlpatterns = patterns('api_manager.system_views',
url(r'^$', 'api_detail'),
url(r'^system$', 'system_detail'),
from api_manager import system_views
urlpatterns = patterns('',
url(r'^$', system_views.ApiDetail.as_view()),
url(r'^system$', system_views.SystemDetail.as_view()),
url(r'^users/*', include('api_manager.users_urls')),
url(r'^groups/*', include('api_manager.groups_urls')),
url(r'^sessions/*', include('api_manager.sessions_urls')),
......
lms/djangoapps/api_manager/users_urls.py
""" Users API URI specification """
from django.conf.urls import patterns, url
urlpatterns = patterns('api_manager.users_views',
url(r'/*$^', 'user_list'),
url(r'^(?P<user_id>[0-9]+)$', 'user_detail'),
url(r'^(?P<user_id>[0-9]+)/courses/*$', 'user_courses_list'),
url(r'^(?P<user_id>[0-9]+)/courses/(?P<course_id>[a-zA-Z0-9/_:]+)$', 'user_courses_detail'),
url(r'^(?P<user_id>[0-9]+)/groups/*$', 'user_groups_list'),
url(r'^(?P<user_id>[0-9]+)/groups/(?P<group_id>[0-9]+)$', 'user_groups_detail'),
)
from rest_framework.urlpatterns import format_suffix_patterns
from api_manager import users_views
urlpatterns = patterns('',
url(r'/*$^', users_views.UsersList.as_view()),
url(r'^(?P<user_id>[0-9]+)$', users_views.UsersDetail.as_view()),
url(r'^(?P<user_id>[0-9]+)/courses/*$', users_views.UsersCoursesList.as_view()),
url(r'^(?P<user_id>[0-9]+)/courses/(?P<course_id>[a-zA-Z0-9/_:]+)$', users_views.UsersCoursesDetail.as_view()),
url(r'^(?P<user_id>[0-9]+)/groups/*$', users_views.UsersGroupsList.as_view()),
url(r'^(?P<user_id>[0-9]+)/groups/(?P<group_id>[0-9]+)$', users_views.UsersGroupsDetail.as_view()),
)
urlpatterns = format_suffix_patterns(urlpatterns)
lms/djangoapps/api_manager/users_views.py
......@@ -12,7 +12,9 @@ from django.utils.translation import ugettext_lazy as _
from rest_framework import status
from rest_framework.decorators import api_view, permission_classes
from rest_framework.response import Response
from rest_framework.views import APIView
from api_manager.models import GroupProfile
from api_manager.permissions import ApiKeyHeaderPermission
from courseware import module_render
from courseware.model_data import FieldDataCache
......@@ -37,7 +39,7 @@ def _generate_base_uri(request):
resource_uri = '{}://{}{}'.format(
protocol,
request.get_host(),
request.path
request.get_full_path()
)
return resource_uri
......@@ -88,83 +90,81 @@ def _save_module_position(request, user, course_id, course_descriptor, position)
saved_module = get_current_child(parent_module)
return saved_module.id
class UsersList(APIView):
permission_classes = (ApiKeyHeaderPermission,)
@api_view(['POST'])
@permission_classes((ApiKeyHeaderPermission,))
def user_list(request):
"""
POST creates a new user in the system
"""
response_data = {}
base_uri = _generate_base_uri(request)
email = request.DATA['email']
username = request.DATA['username']
password = request.DATA['password']
first_name = request.DATA.get('first_name', '')
last_name = request.DATA.get('last_name', '')
# enforce password complexity as an optional feature
if settings.FEATURES.get('ENFORCE_PASSWORD_POLICY', False):
def post(self, request, format=None):
"""
POST creates a new user in the system
"""
response_data = {}
base_uri = _generate_base_uri(request)
email = request.DATA['email']
username = request.DATA['username']
password = request.DATA['password']
first_name = request.DATA.get('first_name', '')
last_name = request.DATA.get('last_name', '')
# enforce password complexity as an optional feature
if settings.FEATURES.get('ENFORCE_PASSWORD_POLICY', False):
try:
validate_password_length(password)
validate_password_complexity(password)
validate_password_dictionary(password)
except ValidationError, err:
status_code = status.HTTP_400_BAD_REQUEST
response_data['message'] = _('Password: ') + '; '.join(err.messages)
return Response(response_data, status=status_code)
try:
validate_password_length(password)
validate_password_complexity(password)
validate_password_dictionary(password)
except ValidationError, err:
validate_email(email)
except ValidationError:
status_code = status.HTTP_400_BAD_REQUEST
response_data['message'] = _('Password: ') + '; '.join(err.messages)
response_data['message'] = _('Valid e-mail is required.')
return Response(response_data, status=status_code)
try:
validate_email(email)
except ValidationError:
status_code = status.HTTP_400_BAD_REQUEST
response_data['message'] = _('Valid e-mail is required.')
return Response(response_data, status=status_code)
try:
validate_slug(username)
except ValidationError:
status_code = status.HTTP_400_BAD_REQUEST
response_data['message'] = _('Username should only consist of A-Z and 0-9, with no spaces.')
return Response(response_data, status=status_code)
try:
validate_slug(username)
except ValidationError:
status_code = status.HTTP_400_BAD_REQUEST
response_data['message'] = _('Username should only consist of A-Z and 0-9, with no spaces.')
return Response(response_data, status=status_code)
try:
user = User.objects.create(email=email, username=username)
except IntegrityError:
user = None
else:
user.set_password(password)
user.first_name = first_name
user.last_name = last_name
user.save()
# add this account creation to password history
# NOTE, this will be a NOP unless the feature has been turned on in configuration
password_history_entry = PasswordHistory()
password_history_entry.create(user)
try:
user = User.objects.create(email=email, username=username)
except IntegrityError:
user = None
else:
user.set_password(password)
user.first_name = first_name
user.last_name = last_name
user.save()
# add this account creation to password history
# NOTE, this will be a NOP unless the feature has been turned on in configuration
password_history_entry = PasswordHistory()
password_history_entry.create(user)
# CDODGE: @TODO: We will have to extend this to look in the CourseEnrollmentAllowed table and
# auto-enroll students when they create a new account. Also be sure to remove from
# the CourseEnrollmentAllow table after the auto-registration has taken place
if user:
status_code = status.HTTP_201_CREATED
response_data = _serialize_user(response_data, user)
response_data['uri'] = '{}/{}'.format(base_uri, str(user.id))
else:
status_code = status.HTTP_409_CONFLICT
response_data['message'] = "User '%s' already exists", username
response_data['field_conflict'] = "username"
return Response(response_data, status=status_code)
# CDODGE: @TODO: We will have to extend this to look in the CourseEnrollmentAllowed table and
# auto-enroll students when they create a new account. Also be sure to remove from
# the CourseEnrollmentAllow table after the auto-registration has taken place
if user:
status_code = status.HTTP_201_CREATED
response_data = _serialize_user(response_data, user)
response_data['uri'] = '{}/{}'.format(base_uri, str(user.id))
else:
status_code = status.HTTP_409_CONFLICT
response_data['message'] = "User '%s' already exists", username
response_data['field_conflict'] = "username"
return Response(response_data, status=status_code)
@api_view(['GET', 'DELETE'])
@permission_classes((ApiKeyHeaderPermission,))
def user_detail(request, user_id):
"""
GET retrieves an existing user from the system
DELETE removes/inactivates/etc. an existing user
"""
if request.method == 'GET':
class UsersDetail(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def get(self, request, user_id, format=None):
"""
GET retrieves an existing user from the system
"""
response_data = {}
base_uri = _generate_base_uri(request)
try:
......@@ -179,7 +179,11 @@ def user_detail(request, user_id):
return Response(response_data, status=status.HTTP_200_OK)
except ObjectDoesNotExist:
return Response(response_data, status=status.HTTP_404_NOT_FOUND)
elif request.method == 'DELETE':
def delete(self, request, user_id, format=None):
"""
DELETE removes/inactivates/etc. an existing user
"""
response_data = {}
try:
existing_user = User.objects.get(id=user_id, is_active=True)
......@@ -191,50 +195,71 @@ def user_detail(request, user_id):
return Response(response_data, status=status.HTTP_204_NO_CONTENT)
@api_view(['POST'])
@permission_classes((ApiKeyHeaderPermission,))
def user_groups_list(request, user_id):
"""
POST creates a new user-group relationship in the system
"""
response_data = {}
group_id = request.DATA['group_id']
base_uri = _generate_base_uri(request)
response_data['uri'] = '{}/{}'.format(base_uri, str(group_id))
try:
existing_user = User.objects.get(id=user_id)
existing_group = Group.objects.get(id=group_id)
except ObjectDoesNotExist:
existing_user = None
existing_group = None
if existing_user and existing_group:
class UsersGroupsList(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def post(self, request, user_id, format=None):
"""
POST creates a new user-group relationship in the system
"""
response_data = {}
group_id = request.DATA['group_id']
base_uri = _generate_base_uri(request)
response_data['uri'] = '{}/{}'.format(base_uri, str(group_id))
try:
existing_relationship = existing_user.groups.get(id=existing_group.id)
existing_user = User.objects.get(id=user_id)
existing_group = Group.objects.get(id=group_id)
except ObjectDoesNotExist:
existing_relationship = None
if existing_relationship is None:
existing_user.groups.add(existing_group.id)
response_data['uri'] = '{}/{}'.format(base_uri, existing_user.id)
response_data['group_id'] = str(existing_group.id)
response_data['user_id'] = str(existing_user.id)
response_status = status.HTTP_201_CREATED
existing_user = None
existing_group = None
if existing_user and existing_group:
try:
existing_relationship = existing_user.groups.get(id=existing_group.id)
except ObjectDoesNotExist:
existing_relationship = None
if existing_relationship is None:
existing_user.groups.add(existing_group.id)
response_data['uri'] = '{}/{}'.format(base_uri, existing_user.id)
response_data['group_id'] = str(existing_group.id)
response_data['user_id'] = str(existing_user.id)
response_status = status.HTTP_201_CREATED
else:
response_data['uri'] = '{}/{}'.format(base_uri, existing_group.id)
response_data['message'] = "Relationship already exists."
response_status = status.HTTP_409_CONFLICT
else:
response_data['uri'] = '{}/{}'.format(base_uri, existing_group.id)
response_data['message'] = "Relationship already exists."
response_status = status.HTTP_409_CONFLICT
else:
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
def get(self, request, user_id, format=None):
"""
GET retrieves the list of groups related to the specified user
"""
try:
existing_user = User.objects.get(id=user_id)
except ObjectDoesNotExist:
return Response({}, status.HTTP_404_NOT_FOUND)
groups = existing_user.groups.all()
response_data = {}
response_data['groups'] = []
for group in groups:
group_profile = GroupProfile.objects.get(group_id=group.id)
group_data = {}
group_data['id'] = group.id
group_data['name'] = group_profile.name
response_data['groups'].append(group_data)
response_status = status.HTTP_200_OK
return Response(response_data, status=response_status)
@api_view(['GET', 'DELETE'])
@permission_classes((ApiKeyHeaderPermission,))
def user_groups_detail(request, user_id, group_id):
"""
GET retrieves an existing user-group relationship from the system
DELETE removes/inactivates/etc. an existing user-group relationship
"""
if request.method == 'GET':
class UsersGroupsDetail(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def get(self, request, user_id, group_id, format=None):
"""
GET retrieves an existing user-group relationship from the system
"""
response_data = {}
base_uri = _generate_base_uri(request)
try:
......@@ -251,21 +276,24 @@ def user_groups_detail(request, user_id, group_id):
else:
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
elif request.method == 'DELETE':
def delete(self, request, user_id, group_id, format=None):
"""
DELETE removes/inactivates/etc. an existing user-group relationship
"""
existing_user = User.objects.get(id=user_id, is_active=True)
existing_user.groups.remove(group_id)
existing_user.save()
return Response({}, status=status.HTTP_204_NO_CONTENT)
@api_view(['POST', 'GET'])
@permission_classes((ApiKeyHeaderPermission,))
def user_courses_list(request, user_id):
"""
POST creates a new course enrollment for a user
GET creates the list of enrolled courses for a user
"""
if request.method == 'POST':
class UsersCoursesList(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def post(self, request, user_id, format=None):
"""
POST creates a new course enrollment for a user
"""
store = modulestore()
response_data = {}
user_id = user_id
......@@ -287,7 +315,11 @@ def user_courses_list(request, user_id):
else:
status_code = status.HTTP_404_NOT_FOUND
return Response(response_data, status=status_code)
elif request.method == 'GET':
def get(self, request, user_id, format=None):
"""
GET creates the list of enrolled courses for a user
"""
store = modulestore()
response_data = []
base_uri = _generate_base_uri(request)
......@@ -312,40 +344,13 @@ def user_courses_list(request, user_id):
return Response(response_data, status=status_code)
@api_view(['GET', 'POST', 'DELETE'])
@permission_classes((ApiKeyHeaderPermission,))
def user_courses_detail(request, user_id, course_id):
"""
GET identifies an ACTIVE course enrollment for the specified user
DELETE unenrolls the specified user from a course
"""
if request.method == 'GET':
store = modulestore()
response_data = {}
base_uri = _generate_base_uri(request)
try:
user = User.objects.get(id=user_id, is_active=True)
course_descriptor = store.get_course(course_id)
except (ObjectDoesNotExist, ValueError):
user = None
course_descriptor = None
if user and CourseEnrollment.is_enrolled(user, course_id):
response_data['user_id'] = user.id
response_data['course_id'] = course_id
response_data['uri'] = base_uri
field_data_cache = FieldDataCache([course_descriptor], course_id, user)
course_module = module_render.get_module(
user,
request,
course_descriptor.location,
field_data_cache,
course_id)
response_data['position'] = course_module.position
response_status = status.HTTP_200_OK
else:
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
elif request.method == 'POST':
class UsersCoursesDetail(APIView):
permission_classes = (ApiKeyHeaderPermission,)
def post(self, request, user_id, course_id, format=None):
"""
POST creates an ACTIVE course enrollment for the specified user
"""
store = modulestore()
base_uri = _generate_base_uri(request)
response_data = {}
......@@ -371,7 +376,41 @@ def user_courses_detail(request, user_id, course_id):
else:
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
elif request.method == 'DELETE':
def get(self, request, user_id, course_id, format=None):
"""
GET identifies an ACTIVE course enrollment for the specified user
"""
store = modulestore()
response_data = {}
base_uri = _generate_base_uri(request)
try:
user = User.objects.get(id=user_id, is_active=True)
course_descriptor = store.get_course(course_id)
except (ObjectDoesNotExist, ValueError):
user = None
course_descriptor = None
if user and CourseEnrollment.is_enrolled(user, course_id):
response_data['user_id'] = user.id
response_data['course_id'] = course_id
response_data['uri'] = base_uri
field_data_cache = FieldDataCache([course_descriptor], course_id, user)
course_module = module_render.get_module(
user,
request,
course_descriptor.location,
field_data_cache,
course_id)
response_data['position'] = course_module.position
response_status = status.HTTP_200_OK
else:
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
def delete(self, request, user_id, course_id, format=None):
"""
DELETE unenrolls the specified user from a course
"""
try:
user = User.objects.get(id=user_id, is_active=True)
except ObjectDoesNotExist:
......
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