Commit 07d7754f by chrisndodge

Merge pull request #14 from edx-solutions/api

Api
parents e2a7b2c5 e1e473bb
"""
Courses API URI specification
The order of the URIs really matters here, due to the slash characters present in the identifiers
"""
from django.conf.urls import patterns, url
urlpatterns = patterns('api_manager.courses_views',
url(r'/*$^', 'courses_list'),
url(r'^(?P<course_id>[a-zA-Z0-9/_:]+)/modules/(?P<module_id>[a-zA-Z0-9/_:]+)/submodules/*$', 'modules_list'),
url(r'^(?P<course_id>[a-zA-Z0-9/_:]+)/modules/(?P<module_id>[a-zA-Z0-9/_:]+)$', 'modules_detail'),
url(r'^(?P<course_id>[a-zA-Z0-9/_:]+)/modules/*$', 'modules_list'),
url(r'^(?P<course_id>[a-zA-Z0-9/_:]+)$', 'courses_detail'),
)
""" API implementation for course-oriented interactions. """
from rest_framework import status
from rest_framework.decorators import api_view, permission_classes
from rest_framework.response import Response
from api_manager.permissions import ApiKeyHeaderPermission
from xmodule.modulestore.django import modulestore
from xmodule.modulestore import Location, InvalidLocationError
def _get_module_submodules(module, submodule_type=None):
"""
Parses the provided module looking for child modules
Matches on submodule type (category) when specified
"""
submodules = []
if hasattr(module, 'children'):
child_modules = module.get_children()
for child_module in child_modules:
if submodule_type:
if getattr(child_module, 'category') == submodule_type:
submodules.append(child_module)
else:
submodules.append(child_module)
return submodules
def _serialize_module(request, course_id, module):
"""
Loads the specified module data into the response dict
This should probably evolve to use DRF serializers
"""
data = {}
if getattr(module, 'id') == course_id:
module_id = module.id
else:
module_id = module.location.url()
data['id'] = module_id
if hasattr(module, 'display_name'):
data['name'] = module.display_name
data['category'] = module.location.category
protocol = 'http'
if request.is_secure():
protocol = protocol + 's'
module_uri = '{}://{}/api/courses/{}'.format(
protocol,
request.get_host(),
course_id.encode('utf-8')
)
# Some things we do only if the module is a course
if (course_id == module_id):
data['number'] = module.location.course
data['org'] = module.location.org
# Other things we do only if the module is not a course
else:
module_uri = '{}/modules/{}'.format(module_uri, module_id)
data['uri'] = module_uri
return data
def _serialize_module_submodules(request, course_id, submodules):
"""
Loads the specified module submodule data into the response dict
This should probably evolve to use DRF serializers
"""
data = []
if submodules:
for submodule in submodules:
submodule_data = _serialize_module(
request,
course_id,
submodule
)
data.append(submodule_data)
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)
# pylint: disable=E1103
"""
Run these tests @ Devstack:
rake fasttest_lms[common/djangoapps/api_manager/tests/test_group_views.py]
"""
import unittest
import uuid
from django.conf import settings
from django.core.cache import cache
from django.test import TestCase, Client
from django.test.utils import override_settings
from courseware.tests.modulestore_config import TEST_DATA_MIXED_MODULESTORE
from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory
TEST_API_KEY = str(uuid.uuid4())
class SecureClient(Client):
""" Django test client using a "secure" connection. """
def __init__(self, *args, **kwargs):
kwargs = kwargs.copy()
kwargs.update({'SERVER_PORT': 443, 'wsgi.url_scheme': 'https'})
super(SecureClient, self).__init__(*args, **kwargs)
@override_settings(MODULESTORE=TEST_DATA_MIXED_MODULESTORE)
@override_settings(EDX_API_KEY=TEST_API_KEY)
class CoursesApiTests(TestCase):
""" Test suite for Courses API views """
def setUp(self):
self.test_server_prefix = 'https://testserver'
self.base_courses_uri = '/api/courses'
self.course = CourseFactory.create()
self.test_data = '<html>{}</html>'.format(str(uuid.uuid4()))
self.chapter = ItemFactory.create(
category="chapter",
parent_location=self.course.location,
data=self.test_data,
display_name="Overview"
)
self.module = ItemFactory.create(
category="videosequence",
parent_location=self.chapter.location,
data=self.test_data,
display_name="Video_Sequence"
)
self.submodule = ItemFactory.create(
category="video",
parent_location=self.module.location,
data=self.test_data,
display_name="Video_Resources"
)
self.test_course_id = self.course.id
self.test_course_name = self.course.display_name
self.test_course_number = self.course.number
self.test_course_org = self.course.org
self.test_chapter_id = self.chapter.id
self.test_module_id = self.module.id
self.test_submodule_id = self.submodule.id
self.base_modules_uri = '/api/courses/' + self.test_course_id + '/modules'
self.base_chapters_uri = self.base_modules_uri + '?type=chapter'
self.client = SecureClient()
cache.clear()
def do_get(self, uri):
"""Submit an HTTP GET request"""
headers = {
'Content-Type': 'application/json',
'X-Edx-Api-Key': str(TEST_API_KEY),
}
print "GET: " + uri
response = self.client.get(uri, headers=headers)
return response
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_course_list_get(self):
test_uri = self.base_courses_uri
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
matched_course = False
for course in response.data:
if matched_course is False and course['id'] == self.test_course_id:
self.assertEqual(course['name'], self.test_course_name)
self.assertEqual(course['number'], self.test_course_number)
self.assertEqual(course['org'], self.test_course_org)
confirm_uri = self.test_server_prefix + test_uri + '/' + course['id']
self.assertEqual(course['uri'], confirm_uri)
matched_course = True
self.assertTrue(matched_course)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_course_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)
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)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_course_detail_get_notfound(self):
test_uri = self.base_courses_uri + '/' + 'p29038cvp9hjwefion'
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_chapter_list_get(self):
test_uri = self.base_chapters_uri
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
matched_chapter = False
for chapter in response.data:
if matched_chapter is False and chapter['id'] == self.test_chapter_id:
self.assertIsNotNone(chapter['uri'])
self.assertGreater(len(chapter['uri']), 0)
confirm_uri = self.test_server_prefix + self.base_modules_uri + '/' + chapter['id']
self.assertEqual(chapter['uri'], confirm_uri)
matched_chapter = True
self.assertTrue(matched_chapter)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_chapter_detail_get(self):
test_uri = self.base_modules_uri + '/' + self.test_chapter_id
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data['id']), 0)
self.assertEqual(response.data['id'], self.test_chapter_id)
confirm_uri = self.test_server_prefix + test_uri
self.assertEqual(response.data['uri'], confirm_uri)
self.assertGreater(len(response.data['modules']), 0)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_modules_list_get(self):
test_uri = self.base_modules_uri + '/' + self.test_module_id
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
matched_submodule = False
for submodule in response.data['modules']:
if matched_submodule is False and submodule['id'] == self.test_submodule_id:
self.assertIsNotNone(submodule['uri'])
self.assertGreater(len(submodule['uri']), 0)
confirm_uri = self.test_server_prefix + self.base_modules_uri + '/' + submodule['id']
self.assertEqual(submodule['uri'], confirm_uri)
matched_submodule = True
self.assertTrue(matched_submodule)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_modules_detail_get(self):
test_uri = self.base_modules_uri + '/' + self.test_module_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_module_id)
confirm_uri = self.test_server_prefix + test_uri
self.assertEqual(response.data['uri'], confirm_uri)
self.assertGreater(len(response.data['modules']), 0)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_modules_detail_get_notfound(self):
test_uri = self.base_modules_uri + '/' + '2p38fp2hjfp9283'
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_modules_list_get_filtered_submodules_for_module(self):
test_uri = self.base_modules_uri + '/' + self.test_module_id + '/submodules?type=video'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
matched_submodule = False
for submodule in response.data:
if matched_submodule is False and submodule['id'] == self.test_submodule_id:
confirm_uri = self.test_server_prefix + self.base_modules_uri + '/' + submodule['id']
self.assertEqual(submodule['uri'], confirm_uri)
matched_submodule = True
self.assertTrue(matched_submodule)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_modules_list_get_notfound(self):
test_uri = self.base_modules_uri + '/2p38fp2hjfp9283/submodules?type=video'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
"""
Courses API URI specification
The order of the URIs really matters here, due to the slash characters present in the identifiers
"""
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>[^/]+/[^/]+/[^/]+)$', 'courses_detail'),
)
""" 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 rest_framework import status
from rest_framework.decorators import api_view, permission_classes
from rest_framework.response import Response
from api_manager.permissions import ApiKeyHeaderPermission
from api_manager.models import CourseGroupRelationship
from xmodule.modulestore.django import modulestore
from xmodule.modulestore import Location, InvalidLocationError
from courseware.courses import get_course_about_section, get_course_info_section
from courseware.views import get_static_tab_contents
from student.models import CourseEnrollment, CourseEnrollmentAllowed
log = logging.getLogger(__name__)
def _generate_base_uri(request):
"""
Constructs the protocol:host:path component of the resource uri
"""
protocol = 'http'
if request.is_secure():
protocol = protocol + 's'
resource_uri = '{}://{}{}'.format(
protocol,
request.get_host(),
request.path
)
return resource_uri
def _get_module_submodules(module, submodule_type=None):
"""
Parses the provided module looking for child modules
Matches on submodule type (category) when specified
"""
submodules = []
if hasattr(module, 'children'):
child_modules = module.get_children()
for child_module in child_modules:
if submodule_type:
if getattr(child_module, 'category') == submodule_type:
submodules.append(child_module)
else:
submodules.append(child_module)
return submodules
def _serialize_module(request, course_id, module):
"""
Loads the specified module data into the response dict
This should probably evolve to use DRF serializers
"""
data = {}
if getattr(module, 'id') == course_id:
module_id = module.id
else:
module_id = module.location.url()
data['id'] = module_id
if hasattr(module, 'display_name'):
data['name'] = module.display_name
data['category'] = module.location.category
protocol = 'http'
if request.is_secure():
protocol = protocol + 's'
module_uri = '{}://{}/api/courses/{}'.format(
protocol,
request.get_host(),
course_id.encode('utf-8')
)
# Some things we do only if the module is a course
if (course_id == module_id):
data['number'] = module.location.course
data['org'] = module.location.org
# Other things we do only if the module is not a course
else:
module_uri = '{}/modules/{}'.format(module_uri, module_id)
data['uri'] = module_uri
return data
def _serialize_module_submodules(request, course_id, submodules):
"""
Loads the specified module submodule data into the response dict
This should probably evolve to use DRF serializers
"""
data = []
if submodules:
for submodule in submodules:
submodule_data = _serialize_module(
request,
course_id,
submodule
)
data.append(submodule_data)
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(['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 _inner_content(tag):
"""
Helper method
"""
inner_content = None
if tag is not None:
inner_content = tag.text if tag.text else u''
inner_content += u''.join(etree.tostring(e) for e in tag)
inner_content += tag.tail if tag.tail else u''
return inner_content
def _parse_overview_html(html):
"""
Helper method to break up the course about HTML into components
"""
result = {}
parser = etree.HTMLParser()
tree = etree.parse(StringIO(html), parser)
sections = tree.findall('/body/section')
result = []
for section in sections:
section_class = section.get('class')
if section_class:
section_data = OrderedDict()
section_data['class'] = section_class
articles = section.findall('article')
if articles:
section_data['articles'] = []
for article in articles:
article_class = article.get('class')
if article_class:
article_data = OrderedDict()
article_data['class'] = article_class
if article_class == "teacher":
name_element = article.find('h3')
if name_element is not None:
article_data['name'] = name_element.text
image_element = article.find("./div[@class='teacher-image']/img")
if image_element is not None:
article_data['image_src'] = image_element.get('src')
bios = article.findall('p')
bio_html = ''
for bio in bios:
bio_html += etree.tostring(bio)
if bio_html:
article_data['bio'] = bio_html
else:
article_data['body'] = _inner_content(article)
section_data['articles'].append(article_data)
else:
section_data['body'] = _inner_content(section)
result.append(section_data)
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
"""
result = {}
parser = etree.HTMLParser()
tree = etree.parse(StringIO(html), parser)
# get all of the individual postings
postings = tree.findall('/body/ol/li')
result = []
for posting in postings:
posting_data = {}
posting_date_element = posting.find('h2')
if posting_date_element is not None:
posting_data['date'] = posting_date_element.text
content = u''
for el in posting:
# note, we can't delete or skip over the date element in
# the HTML tree because there might be some tailing content
if el != posting_date_element:
content += etree.tostring(el)
else:
content += el.tail if el.tail else u''
posting_data['content'] = content.strip()
result.append(posting_data)
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()
try:
course_module = store.get_course(course_id)
if not course_module:
return Response({}, status=status.HTTP_404_NOT_FOUND)
content = get_course_info_section(request, course_module, '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']:
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)
else:
response_data['content'] = content
except InvalidLocationError:
return Response({}, status=status.HTTP_404_NOT_FOUND)
return Response(response_data)
@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:
return Response({}, status=status.HTTP_404_NOT_FOUND)
tabs = []
for tab in course_module.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,
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)
@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()
try:
course_module = store.get_course(course_id)
if not course_module:
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)
@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:
return Response({}, status=status.HTTP_404_NOT_FOUND)
except InvalidLocationError:
return Response({}, status=status.HTTP_404_NOT_FOUND)
if request.method == 'GET':
# Get a list of all enrolled students
users = CourseEnrollment.users_enrolled_in(course_id)
response_data['enrollments'] = []
for user in users:
user_data = OrderedDict()
user_data['id'] = user.id
user_data['email'] = user.email
user_data['username'] = user.username
# @TODO: Should we create a URI resourse that points to user?!? But that's in a different URL subpath
response_data['enrollments'].append(user_data)
# Then list all enrollments which are pending. These are enrollments for students that have not yet
# created an account
pending_enrollments = CourseEnrollmentAllowed.objects.filter(course_id=course_id)
if pending_enrollments:
response_data['pending_enrollments'] = []
for cea in pending_enrollments:
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)
# 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()
return Response({}, status.HTTP_201_CREATED)
...@@ -4,6 +4,8 @@ from django.conf.urls import patterns, url ...@@ -4,6 +4,8 @@ from django.conf.urls import patterns, url
urlpatterns = patterns('api_manager.groups_views', urlpatterns = patterns('api_manager.groups_views',
url(r'/*$^', 'group_list'), url(r'/*$^', 'group_list'),
url(r'^(?P<group_id>[0-9]+)$', 'group_detail'), 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/*$', '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]+)/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/*$', 'group_groups_list'),
......
""" API implementation for gourse-oriented interactions. """ """ API implementation for group-oriented interactions. """
import uuid import uuid
import json
from collections import OrderedDict
from django.contrib.auth.models import Group, User from django.contrib.auth.models import Group, User
from django.core.exceptions import ObjectDoesNotExist from django.core.exceptions import ObjectDoesNotExist
...@@ -10,7 +12,9 @@ from rest_framework.decorators import api_view, permission_classes ...@@ -10,7 +12,9 @@ from rest_framework.decorators import api_view, permission_classes
from rest_framework.response import Response from rest_framework.response import Response
from api_manager.permissions import ApiKeyHeaderPermission from api_manager.permissions import ApiKeyHeaderPermission
from api_manager.models import GroupRelationship from api_manager.models import GroupRelationship, CourseGroupRelationship, GroupProfile
from xmodule.modulestore.django import modulestore
from xmodule.modulestore import Location, InvalidLocationError
RELATIONSHIP_TYPES = {'hierarchical': 'h', 'graph': 'g'} RELATIONSHIP_TYPES = {'hierarchical': 'h', 'graph': 'g'}
...@@ -30,46 +34,73 @@ def _generate_base_uri(request): ...@@ -30,46 +34,73 @@ def _generate_base_uri(request):
return resource_uri return resource_uri
@api_view(['POST']) @api_view(['GET', 'POST'])
@permission_classes((ApiKeyHeaderPermission,)) @permission_classes((ApiKeyHeaderPermission,))
def group_list(request): def group_list(request):
""" """
GET retrieves a list of groups in the system filtered by type
POST creates a new group in the system POST creates a new group in the system
""" """
response_data = {} if request.method == 'GET':
base_uri = _generate_base_uri(request) if not 'type' in request.GET:
# Group name must be unique, but we need to support dupes return Response({}, status=status.HTTP_400_BAD_REQUEST)
group = Group.objects.create(name=str(uuid.uuid4()))
original_group_name = request.DATA['name'] response_data = []
group.name = '{:04d}: {}'.format(group.id, original_group_name) profiles = GroupProfile.objects.filter(group_type=request.GET['type'])
group.record_active = True for profile in profiles:
group.record_date_created = timezone.now() item_data = OrderedDict()
group.record_date_modified = timezone.now() item_data['group_id'] = profile.group_id
group.save() item_data['group_type'] = profile.group_type
# Relationship model also allows us to use duplicate names item_data['data'] = json.loads(profile.data)
GroupRelationship.objects.create(name=original_group_name, group_id=group.id, parent_group=None) response_data.append(item_data)
response_data = {'id': group.id, 'name': original_group_name}
base_uri = _generate_base_uri(request) return Response(response_data)
response_data['uri'] = '{}/{}'.format(base_uri, group.id) elif request.method == 'POST':
response_status = status.HTTP_201_CREATED response_data = {}
return Response(response_data, status=response_status) base_uri = _generate_base_uri(request)
# Group name must be unique, but we need to support dupes
group = Group.objects.create(name=str(uuid.uuid4()))
original_group_name = request.DATA['name']
group.name = '{:04d}: {}'.format(group.id, original_group_name)
group.record_active = True
group.record_date_created = timezone.now()
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)
@api_view(['GET']) # allow for optional meta information about groups, this will end up in the GroupProfile table
group_type = request.DATA.get('group_type')
data = request.DATA.get('data')
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}
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)
@api_view(['GET', 'POST'])
@permission_classes((ApiKeyHeaderPermission,)) @permission_classes((ApiKeyHeaderPermission,))
def group_detail(request, group_id): def group_detail(request, group_id):
""" """
GET retrieves an existing group from the system GET retrieves an existing group from the system
""" """
response_data = {} response_data = {}
base_uri = _generate_base_uri(request) base_uri = _generate_base_uri(request)
try: try:
existing_group = Group.objects.get(id=group_id) existing_group = Group.objects.get(id=group_id)
existing_group_relationship = GroupRelationship.objects.get(group_id=group_id) existing_group_relationship = GroupRelationship.objects.get(group_id=group_id)
except ObjectDoesNotExist: except ObjectDoesNotExist:
existing_group = None return Response({}, status.HTTP_404_NOT_FOUND)
existing_group_relationship = None
if existing_group and existing_group_relationship: if request.method == 'GET':
response_data['name'] = existing_group_relationship.name response_data['name'] = existing_group_relationship.name
response_data['id'] = existing_group.id response_data['id'] = existing_group.id
response_data['uri'] = base_uri response_data['uri'] = base_uri
...@@ -78,10 +109,36 @@ def group_detail(request, group_id): ...@@ -78,10 +109,36 @@ def group_detail(request, group_id):
response_data['resources'].append({'uri': resource_uri}) response_data['resources'].append({'uri': resource_uri})
resource_uri = '{}/groups'.format(base_uri) resource_uri = '{}/groups'.format(base_uri)
response_data['resources'].append({'uri': resource_uri}) response_data['resources'].append({'uri': resource_uri})
# see if there is an (optional) GroupProfile
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)
except ObjectDoesNotExist:
pass
response_status = status.HTTP_200_OK response_status = status.HTTP_200_OK
else:
response_status = status.HTTP_404_NOT_FOUND return Response(response_data, status=response_status)
return Response(response_data, status=response_status) elif request.method == 'POST':
# update GroupProfile data
group_type = request.DATA.get('group_type')
data = request.DATA.get('data')
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()
return Response({})
@api_view(['POST']) @api_view(['POST'])
...@@ -269,3 +326,77 @@ def group_groups_detail(request, group_id, related_group_id): ...@@ -269,3 +326,77 @@ def group_groups_detail(request, group_id, related_group_id):
else: else:
response_status = status.HTTP_404_NOT_FOUND response_status = status.HTTP_404_NOT_FOUND
return Response({}, status=response_status) return Response({}, status=response_status)
@api_view(['POST'])
@permission_classes((ApiKeyHeaderPermission,))
def group_courses_list(request, group_id):
"""
POST creates a new group-course relationship in the system
"""
response_data = {}
group_id = group_id
course_id = request.DATA['course_id']
base_uri = _generate_base_uri(request)
response_data['uri'] = '{}/{}'.format(base_uri, course_id)
store = modulestore()
try:
existing_group = Group.objects.get(id=group_id)
except ObjectDoesNotExist:
existing_group = None
try:
existing_course = store.get_course(course_id)
except ValueError:
existing_course = None
if existing_group and existing_course:
try:
existing_relationship = CourseGroupRelationship.objects.get(course_id=course_id, group=existing_group)
except ObjectDoesNotExist:
existing_relationship = None
if existing_relationship is None:
new_relationship = CourseGroupRelationship.objects.create(course_id=course_id, group=existing_group)
response_data['group_id'] = str(new_relationship.group_id)
response_data['course_id'] = str(new_relationship.course_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 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':
response_data = {}
base_uri = _generate_base_uri(request)
response_data['uri'] = base_uri
try:
existing_group = Group.objects.get(id=group_id)
existing_relationship = CourseGroupRelationship.objects.get(course_id=course_id, group=existing_group)
except ObjectDoesNotExist:
existing_group = None
existing_relationship = None
if existing_group and existing_relationship:
response_data['group_id'] = existing_group.id
response_data['course_id'] = existing_relationship.course_id
response_status = status.HTTP_200_OK
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_group.coursegrouprelationship_set.get(course_id=course_id).delete()
existing_group.save()
except ObjectDoesNotExist:
pass
return Response({}, status=status.HTTP_204_NO_CONTENT)
# -*- 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 model 'CourseGroupRelationship'
db.create_table('api_manager_coursegrouprelationship', (
('id', self.gf('django.db.models.fields.AutoField')(primary_key=True)),
('course_id', self.gf('django.db.models.fields.CharField')(max_length=255, db_index=True)),
('group', self.gf('django.db.models.fields.related.ForeignKey')(to=orm['auth.Group'])),
))
db.send_create_signal('api_manager', ['CourseGroupRelationship'])
# Adding model 'GroupProfile'
db.create_table('auth_groupprofile', (
('id', self.gf('django.db.models.fields.AutoField')(primary_key=True)),
('group', self.gf('django.db.models.fields.related.ForeignKey')(to=orm['auth.Group'])),
('group_type', self.gf('django.db.models.fields.CharField')(max_length=32, null=True, db_index=True)),
('data', self.gf('django.db.models.fields.TextField')(blank=True)),
))
db.send_create_signal('api_manager', ['GroupProfile'])
def backwards(self, orm):
# Deleting model 'CourseGroupRelationship'
db.delete_table('api_manager_coursegrouprelationship')
# Deleting model 'GroupProfile'
db.delete_table('auth_groupprofile')
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'})
},
'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, 21, 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, 21, 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
...@@ -82,3 +82,29 @@ class LinkedGroupRelationship(models.Model): ...@@ -82,3 +82,29 @@ class LinkedGroupRelationship(models.Model):
record_active = models.BooleanField(default=True) record_active = models.BooleanField(default=True)
record_date_created = models.DateTimeField(default=timezone.now()) record_date_created = models.DateTimeField(default=timezone.now())
record_date_modified = models.DateTimeField(auto_now=True) record_date_modified = models.DateTimeField(auto_now=True)
class CourseGroupRelationship(models.Model):
"""
The CourseGroupRelationship model contains information describing the
link between a course and a group. A typical use case for this table
is to manage the courses for an XSeries or other sort of program.
"""
course_id = models.CharField(max_length=255, db_index=True)
group = models.ForeignKey(Group, db_index=True)
class GroupProfile(models.Model):
"""
This table will provide additional tables regarding groups. This has a foreign key to
the auth_groups table
"""
class Meta:
db_table = "auth_groupprofile"
group = models.ForeignKey(Group, db_index=True)
group_type = models.CharField(null=True, max_length=32, db_index=True)
data = models.TextField(blank=True) # JSON dictionary for generic key/value pairs
...@@ -39,10 +39,10 @@ class ApiKeyHeaderPermission(permissions.BasePermission): ...@@ -39,10 +39,10 @@ class ApiKeyHeaderPermission(permissions.BasePermission):
if header_key is None: if header_key is None:
try: try:
header_key = request.META['headers'].get('X-Edx-Api-Key') header_key = request.META['headers'].get('X-Edx-Api-Key')
if header_key is None:
return False
except KeyError: except KeyError:
return False return False
if header_key is None:
return False
# The api key values need to be the same # The api key values need to be the same
if header_key != api_key: if header_key != api_key:
......
"""
Some test content strings. Best to keep them out of the test files because they take up a lot of
text space
"""
from textwrap import dedent
TEST_COURSE_UPDATES_CONTENT = dedent("""
<ol>
<li>
<h2>April 18, 2014</h2>
This does not have a paragraph tag around it
</li>
<li>
<h2>April 17, 2014</h2>
Some text before paragraph tag<p>This is inside paragraph tag</p>Some text after tag
</li>
<li>
<h2>April 16, 2014</h2>
Some text before paragraph tag<p>This is inside paragraph tag</p>Some text after tag<p>one more</p>
</li>
<li>
<h2>April 15, 2014</h2>
<p>A perfectly</p><p>formatted piece</p><p>of HTML</p>
</li>
</ol>
"""
)
TEST_STATIC_TAB1_CONTENT = dedent("""
<div>This is static tab1</div>
"""
)
TEST_STATIC_TAB2_CONTENT = dedent("""
<div>This is static tab2</div>
"""
)
TEST_COURSE_OVERVIEW_CONTENT = dedent("""
<section class="about">
<h2>About This Course</h2>
<p>Include your long course description here. The long course description should contain 150-400 words.</p>
<p>This is paragraph 2 of the long course description. Add more paragraphs as needed. Make sure to enclose them in paragraph tags.</p>
</section>
<section class="prerequisites">
<h2>Prerequisites</h2>
<p>Add information about course prerequisites here.</p>
</section>
<section class="course-staff">
<h2>Course Staff</h2>
<article class="teacher">
<div class="teacher-image">
<img src="/images/pl-faculty.png" align="left" style="margin:0 20 px 0" alt="Course Staff Image #1">
</div>
<h3>Staff Member #1</h3>
<p>Biography of instructor/staff member #1</p>
</article>
<article class="teacher">
<div class="teacher-image">
<img src="/images/pl-faculty.png" align="left" style="margin:0 20 px 0" alt="Course Staff Image #2">
</div>
<h3>Staff Member #2</h3>
<p>Biography of instructor/staff member #2</p>
</article>
</section>
<section class="faq">
<p>Some text here</p>
</section>
""")
# pylint: disable=E1103
"""
Run these tests @ Devstack:
rake fasttest_lms[common/djangoapps/api_manager/tests/test_group_views.py]
"""
import simplejson as json
import unittest
import uuid
from random import randint
from django.core.cache import cache
from django.test import TestCase, Client
from django.test.utils import override_settings
from courseware.tests.modulestore_config import TEST_DATA_MIXED_MODULESTORE
from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory
from .content import TEST_COURSE_OVERVIEW_CONTENT, TEST_COURSE_UPDATES_CONTENT
from .content import TEST_STATIC_TAB1_CONTENT, TEST_STATIC_TAB2_CONTENT
TEST_API_KEY = str(uuid.uuid4())
class SecureClient(Client):
""" Django test client using a "secure" connection. """
def __init__(self, *args, **kwargs):
kwargs = kwargs.copy()
kwargs.update({'SERVER_PORT': 443, 'wsgi.url_scheme': 'https'})
super(SecureClient, self).__init__(*args, **kwargs)
@override_settings(MODULESTORE=TEST_DATA_MIXED_MODULESTORE)
@override_settings(EDX_API_KEY=TEST_API_KEY)
class CoursesApiTests(TestCase):
""" Test suite for Courses API views """
def setUp(self):
self.maxDiff = 3000
self.test_server_prefix = 'https://testserver'
self.base_courses_uri = '/api/courses'
self.base_groups_uri = '/api/groups'
self.test_group_name = 'Alpha Group'
self.course = CourseFactory.create()
self.test_data = '<html>{}</html>'.format(str(uuid.uuid4()))
self.chapter = ItemFactory.create(
category="chapter",
parent_location=self.course.location,
data=self.test_data,
display_name="Overview"
)
self.module = ItemFactory.create(
category="videosequence",
parent_location=self.chapter.location,
data=self.test_data,
display_name="Video_Sequence"
)
self.submodule = ItemFactory.create(
category="video",
parent_location=self.module.location,
data=self.test_data,
display_name="Video_Resources"
)
self.overview = ItemFactory.create(
category="about",
parent_location=self.course.location,
data=TEST_COURSE_OVERVIEW_CONTENT,
display_name="overview"
)
self.updates = ItemFactory.create(
category="course_info",
parent_location=self.course.location,
data=TEST_COURSE_UPDATES_CONTENT,
display_name="updates"
)
self.static_tab1 = ItemFactory.create(
category="static_tab",
parent_location=self.course.location,
data=TEST_STATIC_TAB1_CONTENT,
display_name="syllabus"
)
self.static_tab2 = ItemFactory.create(
category="static_tab",
parent_location=self.course.location,
data=TEST_STATIC_TAB2_CONTENT,
display_name="readings"
)
self.test_course_id = self.course.id
self.test_bogus_course_id = 'foo/bar/baz'
self.test_course_name = self.course.display_name
self.test_course_number = self.course.number
self.test_course_org = self.course.org
self.test_chapter_id = self.chapter.id
self.test_module_id = self.module.id
self.test_submodule_id = self.submodule.id
self.base_modules_uri = '/api/courses/' + self.test_course_id + '/modules'
self.base_chapters_uri = self.base_modules_uri + '?type=chapter'
self.client = SecureClient()
cache.clear()
def do_get(self, uri):
"""Submit an HTTP GET request"""
headers = {
'Content-Type': 'application/json',
'X-Edx-Api-Key': str(TEST_API_KEY),
}
response = self.client.get(uri, headers=headers)
return response
def do_post(self, uri, data):
"""Submit an HTTP POST request"""
headers = {
'X-Edx-Api-Key': str(TEST_API_KEY),
}
json_data = json.dumps(data)
response = self.client.post(uri, headers=headers, content_type='application/json', data=json_data)
return response
def do_delete(self, uri):
"""Submit an HTTP DELETE request"""
headers = {
'Content-Type': 'application/json',
'X-Edx-Api-Key': str(TEST_API_KEY),
}
response = self.client.delete(uri, headers=headers)
return response
def test_course_list_get(self):
test_uri = self.base_courses_uri
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
matched_course = False
for course in response.data:
if matched_course is False and course['id'] == self.test_course_id:
self.assertEqual(course['name'], self.test_course_name)
self.assertEqual(course['number'], self.test_course_number)
self.assertEqual(course['org'], self.test_course_org)
confirm_uri = self.test_server_prefix + test_uri + '/' + course['id']
self.assertEqual(course['uri'], confirm_uri)
matched_course = True
self.assertTrue(matched_course)
def test_course_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)
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'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_chapter_list_get(self):
test_uri = self.base_chapters_uri
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
matched_chapter = False
for chapter in response.data:
if matched_chapter is False and chapter['id'] == self.test_chapter_id:
self.assertIsNotNone(chapter['uri'])
self.assertGreater(len(chapter['uri']), 0)
confirm_uri = self.test_server_prefix + self.base_modules_uri + '/' + chapter['id']
self.assertEqual(chapter['uri'], confirm_uri)
matched_chapter = True
self.assertTrue(matched_chapter)
def test_chapter_detail_get(self):
test_uri = self.base_modules_uri + '/' + self.test_chapter_id
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data['id']), 0)
self.assertEqual(response.data['id'], self.test_chapter_id)
confirm_uri = self.test_server_prefix + test_uri
self.assertEqual(response.data['uri'], confirm_uri)
self.assertGreater(len(response.data['modules']), 0)
def test_modules_list_get(self):
test_uri = self.base_modules_uri + '/' + self.test_module_id
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
matched_submodule = False
for submodule in response.data['modules']:
if matched_submodule is False and submodule['id'] == self.test_submodule_id:
self.assertIsNotNone(submodule['uri'])
self.assertGreater(len(submodule['uri']), 0)
confirm_uri = self.test_server_prefix + self.base_modules_uri + '/' + submodule['id']
self.assertEqual(submodule['uri'], confirm_uri)
matched_submodule = True
self.assertTrue(matched_submodule)
def test_modules_detail_get(self):
test_uri = self.base_modules_uri + '/' + self.test_module_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_module_id)
confirm_uri = self.test_server_prefix + test_uri
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)
self.assertEqual(response.status_code, 404)
def test_modules_list_get_filtered_submodules_for_module(self):
test_uri = self.base_modules_uri + '/' + self.test_module_id + '/submodules?type=video'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
matched_submodule = False
for submodule in response.data:
if matched_submodule is False and submodule['id'] == self.test_submodule_id:
confirm_uri = self.test_server_prefix + self.base_modules_uri + '/' + submodule['id']
self.assertEqual(submodule['uri'], confirm_uri)
matched_submodule = True
self.assertTrue(matched_submodule)
def test_modules_list_get_notfound(self):
test_uri = self.base_modules_uri + '/2p38fp2hjfp9283/submodules?type=video'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_course_groups_list_post(self):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
group_id = response.data['id']
test_uri = '{}/{}/groups'.format(self.base_courses_uri, self.test_course_id)
data = {'group_id': group_id}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 201)
confirm_uri = self.test_server_prefix + test_uri + '/' + str(group_id)
self.assertEqual(response.data['uri'], confirm_uri)
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):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
group_id = response.data['id']
test_uri = '{}/{}/groups'.format(self.base_courses_uri, self.test_course_id)
data = {'group_id': group_id}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 201)
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'
data = {'group_id': "98723896"}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 404)
def test_course_groups_detail_get(self):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
group_id = response.data['id']
test_uri = '{}/{}/groups'.format(self.base_courses_uri, self.test_course_id)
data = {'group_id': response.data['id']}
response = self.do_post(test_uri, data)
test_uri = response.data['uri']
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data['uri'], test_uri)
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):
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)
data = {'group_id': response.data['id']}
response = self.do_post(test_uri, data)
test_uri = response.data['uri']
response = self.do_delete(test_uri)
self.assertEqual(response.status_code, 204)
response = self.do_delete(test_uri)
self.assertEqual(response.status_code, 204) # Idempotent
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_course_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):
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):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
group_id = response.data['id']
test_uri = '{}/{}/groups/{}'.format(self.base_courses_uri, self.test_course_id, group_id)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_get_course_overview_unparsed(self):
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/overview'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
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):
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)
self.assertGreater(len(response.data), 0)
sections = response.data['sections']
self.assertEqual(len(sections), 4)
self.assertIsNotNone(self._find_item_by_class(sections, 'about'))
self.assertIsNotNone(self._find_item_by_class(sections, 'prerequisites'))
self.assertIsNotNone(self._find_item_by_class(sections, 'course-staff'))
self.assertIsNotNone(self._find_item_by_class(sections, 'faq'))
course_staff = self._find_item_by_class(sections, 'course-staff')
teachers = course_staff['articles']
self.assertEqual(len(teachers), 2)
self.assertEqual(teachers[0]['name'], "Staff Member #1")
self.assertEqual(teachers[0]['image_src'], "/images/pl-faculty.png")
self.assertIn("<p>Biography of instructor/staff member #1</p>", teachers[0]['bio'])
self.assertEqual(teachers[1]['name'], "Staff Member #2")
self.assertEqual(teachers[1]['image_src'], "/images/pl-faculty.png")
self.assertIn("<p>Biography of instructor/staff member #2</p>", teachers[1]['bio'])
about = self._find_item_by_class(sections, 'about')
self.assertGreater(len(about['body']), 0)
prerequisites = self._find_item_by_class(sections, 'prerequisites')
self.assertGreater(len(prerequisites['body']), 0)
faq = self._find_item_by_class(sections, 'faq')
self.assertGreater(len(faq['body']), 0)
def test_get_course_updates(self):
# first try raw without any parsing
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/updates'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
self.assertEqual(response.data['content'], self.updates.data)
# then try parsed
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/updates?parse=True'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
postings = response.data['postings']
self.assertEqual(len(postings), 4)
self.assertEqual(postings[0]['date'], 'April 18, 2014')
self.assertEqual(postings[0]['content'], 'This does not have a paragraph tag around it')
self.assertEqual(postings[1]['date'], 'April 17, 2014')
self.assertEqual(postings[1]['content'], 'Some text before paragraph tag<p>This is inside paragraph tag</p>Some text after tag')
self.assertEqual(postings[2]['date'], 'April 16, 2014')
self.assertEqual(postings[2]['content'], 'Some text before paragraph tag<p>This is inside paragraph tag</p>Some text after tag<p>one more</p>')
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'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
tabs = response.data['tabs']
self.assertEqual(len(tabs), 2)
self.assertEqual(tabs[0]['name'], u'syllabus')
self.assertEqual(tabs[0]['id'], u'syllabus')
self.assertEqual(tabs[1]['name'], u'readings')
self.assertEqual(tabs[1]['id'], u'readings')
# now try when we get the details on the tabs
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/static_tabs?detail=true'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
tabs = response.data['tabs']
self.assertEqual(tabs[0]['name'], u'syllabus')
self.assertEqual(tabs[0]['id'], u'syllabus')
self.assertEqual(tabs[0]['content'], self.static_tab1.data)
self.assertEqual(tabs[1]['name'], u'readings')
self.assertEqual(tabs[1]['id'], u'readings')
self.assertEqual(tabs[1]['content'], self.static_tab2.data)
#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):
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)
self.assertGreater(len(response.data), 0)
tab = response.data
self.assertEqual(tab['name'], u'syllabus')
self.assertEqual(tab['id'], u'syllabus')
self.assertEqual(tab['content'], self.static_tab1.data)
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/static_tabs/readings'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
tab = response.data
self.assertEqual(tab['name'], u'readings')
self.assertEqual(tab['id'], u'readings')
self.assertEqual(tab['content'], self.static_tab2.data)
# 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)
# 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):
test_uri = self.base_courses_uri + '/' + self.test_course_id + '/users'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertGreater(len(response.data), 0)
# assert that there is no enrolled students
enrollments = response.data['enrollments']
self.assertEqual(len(enrollments), 0)
self.assertNotIn('pending_enrollments', response.data)
# 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
response = self.do_post(test_uri, post_data)
self.assertEqual(response.status_code, 400)
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)
# 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')
# create a new user (note, this calls into the /users/ subsystem)
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 register this user
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
response = self.do_get(test_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)
...@@ -5,15 +5,16 @@ Run these tests @ Devstack: ...@@ -5,15 +5,16 @@ Run these tests @ Devstack:
rake fasttest_lms[common/djangoapps/api_manager/tests/test_group_views.py] rake fasttest_lms[common/djangoapps/api_manager/tests/test_group_views.py]
""" """
from random import randint from random import randint
import unittest
import uuid import uuid
import json
from django.conf import settings
from django.core.cache import cache from django.core.cache import cache
from django.test import TestCase, Client from django.test import TestCase, Client
from django.test.utils import override_settings from django.test.utils import override_settings
from api_manager.models import GroupRelationship from api_manager.models import GroupRelationship
from courseware.tests.modulestore_config import TEST_DATA_MIXED_MODULESTORE
from xmodule.modulestore.tests.factories import CourseFactory
TEST_API_KEY = str(uuid.uuid4()) TEST_API_KEY = str(uuid.uuid4())
...@@ -26,6 +27,7 @@ class SecureClient(Client): ...@@ -26,6 +27,7 @@ class SecureClient(Client):
super(SecureClient, self).__init__(*args, **kwargs) super(SecureClient, self).__init__(*args, **kwargs)
@override_settings(MODULESTORE=TEST_DATA_MIXED_MODULESTORE)
@override_settings(EDX_API_KEY=TEST_API_KEY) @override_settings(EDX_API_KEY=TEST_API_KEY)
class GroupsApiTests(TestCase): class GroupsApiTests(TestCase):
""" Test suite for Groups API views """ """ Test suite for Groups API views """
...@@ -39,6 +41,9 @@ class GroupsApiTests(TestCase): ...@@ -39,6 +41,9 @@ class GroupsApiTests(TestCase):
self.base_users_uri = '/api/users' self.base_users_uri = '/api/users'
self.base_groups_uri = '/api/groups' self.base_groups_uri = '/api/groups'
self.course = CourseFactory.create()
self.test_course_id = self.course.id
self.client = SecureClient() self.client = SecureClient()
cache.clear() cache.clear()
...@@ -69,7 +74,6 @@ class GroupsApiTests(TestCase): ...@@ -69,7 +74,6 @@ class GroupsApiTests(TestCase):
response = self.client.delete(uri, headers=headers) response = self.client.delete(uri, headers=headers)
return response return response
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_list_post(self): def test_group_list_post(self):
data = {'name': self.test_group_name} data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data) response = self.do_post(self.base_groups_uri, data)
...@@ -90,7 +94,6 @@ class GroupsApiTests(TestCase): ...@@ -90,7 +94,6 @@ class GroupsApiTests(TestCase):
# response = self.do_post(self.base_groups_uri, data) # response = self.do_post(self.base_groups_uri, data)
# self.assertEqual(response.status_code, 409) # self.assertEqual(response.status_code, 409)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_detail_get(self): def test_group_detail_get(self):
data = {'name': self.test_group_name} data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data) response = self.do_post(self.base_groups_uri, data)
...@@ -104,13 +107,74 @@ class GroupsApiTests(TestCase): ...@@ -104,13 +107,74 @@ class GroupsApiTests(TestCase):
self.assertEqual(response.data['uri'], confirm_uri) self.assertEqual(response.data['uri'], confirm_uri)
self.assertEqual(response.data['name'], self.test_group_name) self.assertEqual(response.data['name'], self.test_group_name)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms') def test_group_list_get_with_profile(self):
data = {
'name': self.test_group_name,
'group_type': 'series',
'data': json.dumps({'display_name': 'My first series'})
}
response = self.do_post(self.base_groups_uri, data)
self.assertGreater(response.data['id'], 0)
group_id = response.data['id']
# query for list of groups, but don't put the type filter (bad)
test_uri = self.base_groups_uri
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 400)
# try again with filter
test_uri = self.base_groups_uri + '?type=series'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
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]['data']['display_name'], 'My first series')
# query the group detail
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)
self.assertEqual(response.data['group_type'], 'series')
self.assertEqual(response.data['data']['display_name'], 'My first series')
# update the profile
# first with missing data
response = self.do_post(test_uri, {})
self.assertEqual(response.status_code, 400)
data = {
'name': self.test_group_name,
'group_type': 'seriesX',
'data': json.dumps({'display_name': 'My updated series'})
}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 200)
# requery the filter
test_uri = self.base_groups_uri + '?type=series'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data), 0)
test_uri = self.base_groups_uri + '?type=seriesX'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
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]['data']['display_name'], 'My updated series')
def test_group_detail_get_undefined(self): def test_group_detail_get_undefined(self):
test_uri = self.base_groups_uri + '/123456789' test_uri = self.base_groups_uri + '/123456789'
response = self.do_get(test_uri) response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404) self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_users_list_post(self): def test_group_users_list_post(self):
local_username = self.test_username + str(randint(11, 99)) local_username = self.test_username + str(randint(11, 99))
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password} data = {'email': self.test_email, 'username': local_username, 'password': self.test_password}
...@@ -130,7 +194,6 @@ class GroupsApiTests(TestCase): ...@@ -130,7 +194,6 @@ class GroupsApiTests(TestCase):
self.assertEqual(response.data['group_id'], str(group_id)) self.assertEqual(response.data['group_id'], str(group_id))
self.assertEqual(response.data['user_id'], str(user_id)) self.assertEqual(response.data['user_id'], str(user_id))
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_users_list_post_duplicate(self): def test_group_users_list_post_duplicate(self):
local_username = self.test_username + str(randint(11, 99)) local_username = self.test_username + str(randint(11, 99))
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password} data = {'email': self.test_email, 'username': local_username, 'password': self.test_password}
...@@ -147,7 +210,6 @@ class GroupsApiTests(TestCase): ...@@ -147,7 +210,6 @@ class GroupsApiTests(TestCase):
response = self.do_post(test_uri, data) response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 409) self.assertEqual(response.status_code, 409)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_users_list_post_invalid_resources(self): def test_group_users_list_post_invalid_resources(self):
test_uri = self.base_groups_uri + '/1239878976' test_uri = self.base_groups_uri + '/1239878976'
test_uri = test_uri + '/users' test_uri = test_uri + '/users'
...@@ -155,7 +217,6 @@ class GroupsApiTests(TestCase): ...@@ -155,7 +217,6 @@ class GroupsApiTests(TestCase):
response = self.do_post(test_uri, data) response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 404) self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_users_detail_get(self): def test_group_users_detail_get(self):
local_username = self.test_username + str(randint(11, 99)) local_username = self.test_username + str(randint(11, 99))
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password} data = {'email': self.test_email, 'username': local_username, 'password': self.test_password}
...@@ -178,7 +239,6 @@ class GroupsApiTests(TestCase): ...@@ -178,7 +239,6 @@ class GroupsApiTests(TestCase):
self.assertEqual(response.data['group_id'], group_id) self.assertEqual(response.data['group_id'], group_id)
self.assertEqual(response.data['user_id'], user_id) self.assertEqual(response.data['user_id'], user_id)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_users_detail_delete(self): def test_group_users_detail_delete(self):
local_username = self.test_username + str(randint(11, 99)) local_username = self.test_username + str(randint(11, 99))
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password} data = {'email': self.test_email, 'username': local_username, 'password': self.test_password}
...@@ -199,13 +259,11 @@ class GroupsApiTests(TestCase): ...@@ -199,13 +259,11 @@ class GroupsApiTests(TestCase):
response = self.do_get(test_uri) response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404) self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_users_detail_delete_invalid_group(self): def test_group_users_detail_delete_invalid_group(self):
test_uri = self.base_groups_uri + '/123987102/users/123124' test_uri = self.base_groups_uri + '/123987102/users/123124'
response = self.do_delete(test_uri) response = self.do_delete(test_uri)
self.assertEqual(response.status_code, 204) self.assertEqual(response.status_code, 204)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_users_detail_delete_invalid_user(self): def test_group_users_detail_delete_invalid_user(self):
data = {'name': self.test_group_name} data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data) response = self.do_post(self.base_groups_uri, data)
...@@ -214,7 +272,6 @@ class GroupsApiTests(TestCase): ...@@ -214,7 +272,6 @@ class GroupsApiTests(TestCase):
response = self.do_delete(test_uri) response = self.do_delete(test_uri)
self.assertEqual(response.status_code, 204) self.assertEqual(response.status_code, 204)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_users_detail_get_undefined(self): def test_group_users_detail_get_undefined(self):
local_username = self.test_username + str(randint(11, 99)) local_username = self.test_username + str(randint(11, 99))
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password} data = {'email': self.test_email, 'username': local_username, 'password': self.test_password}
...@@ -227,7 +284,6 @@ class GroupsApiTests(TestCase): ...@@ -227,7 +284,6 @@ class GroupsApiTests(TestCase):
response = self.do_get(test_uri) response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404) self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_groups_list_post_hierarchical(self): def test_group_groups_list_post_hierarchical(self):
data = {'name': 'Alpha Group'} data = {'name': 'Alpha Group'}
alpha_response = self.do_post(self.base_groups_uri, data) alpha_response = self.do_post(self.base_groups_uri, data)
...@@ -250,7 +306,6 @@ class GroupsApiTests(TestCase): ...@@ -250,7 +306,6 @@ class GroupsApiTests(TestCase):
self.assertEqual(response.data['group_id'], str(group_id)) self.assertEqual(response.data['group_id'], str(group_id))
self.assertEqual(response.data['relationship_type'], relationship_type) self.assertEqual(response.data['relationship_type'], relationship_type)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_groups_list_post_linked(self): def test_group_groups_list_post_linked(self):
data = {'name': 'Alpha Group'} data = {'name': 'Alpha Group'}
alpha_response = self.do_post(self.base_groups_uri, data) alpha_response = self.do_post(self.base_groups_uri, data)
...@@ -273,7 +328,6 @@ class GroupsApiTests(TestCase): ...@@ -273,7 +328,6 @@ class GroupsApiTests(TestCase):
self.assertEqual(response.data['group_id'], str(group_id)) self.assertEqual(response.data['group_id'], str(group_id))
self.assertEqual(response.data['relationship_type'], relationship_type) self.assertEqual(response.data['relationship_type'], relationship_type)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_groups_list_post_linked_duplicate(self): def test_group_groups_list_post_linked_duplicate(self):
data = {'name': 'Alpha Group'} data = {'name': 'Alpha Group'}
alpha_response = self.do_post(self.base_groups_uri, data) alpha_response = self.do_post(self.base_groups_uri, data)
...@@ -294,7 +348,6 @@ class GroupsApiTests(TestCase): ...@@ -294,7 +348,6 @@ class GroupsApiTests(TestCase):
# Duplicate responses are idemnotent in this case # Duplicate responses are idemnotent in this case
self.assertEqual(response.status_code, 201) self.assertEqual(response.status_code, 201)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_groups_list_post_invalid_group(self): def test_group_groups_list_post_invalid_group(self):
test_uri = self.base_groups_uri + '/123098/groups' test_uri = self.base_groups_uri + '/123098/groups'
relationship_type = 'g' # Graph relationship_type = 'g' # Graph
...@@ -302,7 +355,6 @@ class GroupsApiTests(TestCase): ...@@ -302,7 +355,6 @@ class GroupsApiTests(TestCase):
response = self.do_post(test_uri, data) response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 404) self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_groups_list_post_invalid_relationship_type(self): def test_group_groups_list_post_invalid_relationship_type(self):
data = {'name': 'Alpha Group'} data = {'name': 'Alpha Group'}
alpha_response = self.do_post(self.base_groups_uri, data) alpha_response = self.do_post(self.base_groups_uri, data)
...@@ -320,7 +372,6 @@ class GroupsApiTests(TestCase): ...@@ -320,7 +372,6 @@ class GroupsApiTests(TestCase):
response = self.do_post(test_uri, data) response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 406) self.assertEqual(response.status_code, 406)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_groups_list_get(self): def test_group_groups_list_get(self):
data = {'name': 'Bravo Group'} data = {'name': 'Bravo Group'}
bravo_response = self.do_post(self.base_groups_uri, data) bravo_response = self.do_post(self.base_groups_uri, data)
...@@ -370,13 +421,11 @@ class GroupsApiTests(TestCase): ...@@ -370,13 +421,11 @@ class GroupsApiTests(TestCase):
self.assertGreater(len(relationship['uri']), 0) self.assertGreater(len(relationship['uri']), 0)
self.assertEqual(relationship_count, len(group_idlist)) self.assertEqual(relationship_count, len(group_idlist))
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_groups_list_get_notfound(self): def test_group_groups_list_get_notfound(self):
test_uri = self.base_groups_uri + '/213213123/groups' test_uri = self.base_groups_uri + '/213213123/groups'
response = self.do_get(test_uri) response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404) self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_groups_detail_get_hierarchical(self): def test_group_groups_detail_get_hierarchical(self):
data = {'name': 'Alpha Group'} data = {'name': 'Alpha Group'}
alpha_response = self.do_post(self.base_groups_uri, data) alpha_response = self.do_post(self.base_groups_uri, data)
...@@ -404,7 +453,6 @@ class GroupsApiTests(TestCase): ...@@ -404,7 +453,6 @@ class GroupsApiTests(TestCase):
self.assertEqual(response.data['to_group_id'], str(delta_group_id)) self.assertEqual(response.data['to_group_id'], str(delta_group_id))
self.assertEqual(response.data['relationship_type'], relationship_type) self.assertEqual(response.data['relationship_type'], relationship_type)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_groups_detail_get_linked(self): def test_group_groups_detail_get_linked(self):
data = {'name': 'Alpha Group'} data = {'name': 'Alpha Group'}
alpha_response = self.do_post(self.base_groups_uri, data) alpha_response = self.do_post(self.base_groups_uri, data)
...@@ -435,7 +483,6 @@ class GroupsApiTests(TestCase): ...@@ -435,7 +483,6 @@ class GroupsApiTests(TestCase):
self.assertEqual(response.data['to_group_id'], str(delta_group_id)) self.assertEqual(response.data['to_group_id'], str(delta_group_id))
self.assertEqual(response.data['relationship_type'], relationship_type) self.assertEqual(response.data['relationship_type'], relationship_type)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_groups_detail_get_notfound(self): def test_group_groups_detail_get_notfound(self):
data = {'name': 'Alpha Group'} data = {'name': 'Alpha Group'}
alpha_response = self.do_post(self.base_groups_uri, data) alpha_response = self.do_post(self.base_groups_uri, data)
...@@ -444,7 +491,6 @@ class GroupsApiTests(TestCase): ...@@ -444,7 +491,6 @@ class GroupsApiTests(TestCase):
response = self.do_get(test_uri) response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404) self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_groups_detail_delete_hierarchical(self): def test_group_groups_detail_delete_hierarchical(self):
data = {'name': 'Alpha Group'} data = {'name': 'Alpha Group'}
alpha_response = self.do_post(self.base_groups_uri, data) alpha_response = self.do_post(self.base_groups_uri, data)
...@@ -476,7 +522,6 @@ class GroupsApiTests(TestCase): ...@@ -476,7 +522,6 @@ class GroupsApiTests(TestCase):
response = self.do_get(test_uri) response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404) self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_groups_detail_delete_linked(self): def test_group_groups_detail_delete_linked(self):
data = {'name': 'Alpha Group'} data = {'name': 'Alpha Group'}
alpha_response = self.do_post(self.base_groups_uri, data) alpha_response = self.do_post(self.base_groups_uri, data)
...@@ -506,8 +551,99 @@ class GroupsApiTests(TestCase): ...@@ -506,8 +551,99 @@ class GroupsApiTests(TestCase):
response = self.do_get(test_uri) response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404) self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_group_groups_detail_delete_invalid(self): def test_group_groups_detail_delete_invalid(self):
test_uri = self.base_groups_uri + '/1231234232/groups/1' test_uri = self.base_groups_uri + '/1231234232/groups/1'
response = self.do_delete(test_uri) response = self.do_delete(test_uri)
self.assertEqual(response.status_code, 404) self.assertEqual(response.status_code, 404)
def test_group_courses_list_post(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)
def test_group_courses_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)
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)
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_groups_uri + '/1239878976/courses'
data = {'course_id': "98723896"}
response = self.do_post(test_uri, data)
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)
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)
test_uri = '{}/{}/courses/{}'.format(self.base_groups_uri, group_id, self.test_course_id)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
confirm_uri = '{}{}/{}/courses/{}'.format(
self.test_server_prefix,
self.base_groups_uri,
group_id,
self.test_course_id
)
self.assertEqual(response.data['uri'], confirm_uri)
self.assertEqual(response.data['group_id'], group_id)
self.assertEqual(response.data['course_id'], self.test_course_id)
def test_group_courses_detail_delete(self):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
self.assertEqual(response.status_code, 201)
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)
test_uri = response.data['uri']
response = self.do_delete(test_uri)
self.assertEqual(response.status_code, 204)
response = self.do_delete(test_uri)
self.assertEqual(response.status_code, 204) # Idempotent
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
def test_group_courses_detail_delete_invalid_group(self):
test_uri = self.base_groups_uri + '/123987102/courses/123124'
response = self.do_delete(test_uri)
self.assertEqual(response.status_code, 204)
def test_group_courses_detail_delete_invalid_course(self):
data = {'name': self.test_group_name}
response = self.do_post(self.base_groups_uri, data)
self.assertEqual(response.status_code, 201)
test_uri = response.data['uri'] + '/courses/123124'
response = self.do_delete(test_uri)
self.assertEqual(response.status_code, 204)
def test_group_courses_detail_get_undefined(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 = '{}/courses/{}'.format(response.data['uri'], self.course.id)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
...@@ -51,9 +51,6 @@ class UsersApiTests(TestCase): ...@@ -51,9 +51,6 @@ class UsersApiTests(TestCase):
'X-Edx-Api-Key': str(TEST_API_KEY), 'X-Edx-Api-Key': str(TEST_API_KEY),
} }
json_data = json.dumps(data) json_data = json.dumps(data)
print "POST: " + uri
print json_data
print ""
response = self.client.post(uri, headers=headers, content_type='application/json', data=json_data) response = self.client.post(uri, headers=headers, content_type='application/json', data=json_data)
return response return response
......
...@@ -105,6 +105,9 @@ def user_list(request): ...@@ -105,6 +105,9 @@ def user_list(request):
user.last_name = last_name user.last_name = last_name
user.save() user.save()
# 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: if user:
status_code = status.HTTP_201_CREATED status_code = status.HTTP_201_CREATED
response_data = _serialize_user(response_data, user) response_data = _serialize_user(response_data, user)
......
...@@ -773,7 +773,7 @@ def notification_image_for_tab(course_tab, user, course): ...@@ -773,7 +773,7 @@ def notification_image_for_tab(course_tab, user, course):
return None return None
def get_static_tab_contents(request, course, tab): def get_static_tab_contents(request, course, tab, wrap_xmodule_display=True):
""" """
Returns the contents for the given static tab Returns the contents for the given static tab
""" """
...@@ -788,7 +788,7 @@ def get_static_tab_contents(request, course, tab): ...@@ -788,7 +788,7 @@ def get_static_tab_contents(request, course, tab):
course.id, request.user, modulestore().get_instance(course.id, loc), depth=0 course.id, request.user, modulestore().get_instance(course.id, loc), depth=0
) )
tab_module = get_module( tab_module = get_module(
request.user, request, loc, field_data_cache, course.id, static_asset_path=course.static_asset_path request.user, request, loc, field_data_cache, course.id, static_asset_path=course.static_asset_path, wrap_xmodule_display=wrap_xmodule_display
) )
logging.debug('course_module = {0}'.format(tab_module)) logging.debug('course_module = {0}'.format(tab_module))
......
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