Commit fff9d6af by Matt Drayer Committed by Xavier Antoviaque

Initial API implementation

parent a0c036da
"""
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)
""" Groups API URI specification """
from django.conf.urls import patterns, url
urlpatterns = patterns('api_manager.groups_views',
url(r'/*$^', 'group_list'),
url(r'^(?P<group_id>[0-9]+)$', 'group_detail'),
url(r'^(?P<group_id>[0-9]+)/users/*$', 'group_users_list'),
url(r'^(?P<group_id>[0-9]+)/users/(?P<user_id>[0-9]+)$', 'group_users_detail'),
url(r'^(?P<group_id>[0-9]+)/groups/*$', 'group_groups_list'),
url(r'^(?P<group_id>[0-9]+)/groups/(?P<related_group_id>[0-9]+)$', 'group_groups_detail'),
)
# -*- 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 'GroupRelationship'
db.create_table('api_manager_grouprelationship', (
('group', self.gf('django.db.models.fields.related.OneToOneField')(to=orm['auth.Group'], unique=True, primary_key=True)),
('name', self.gf('django.db.models.fields.CharField')(max_length=255)),
('parent_group', self.gf('django.db.models.fields.related.ForeignKey')(default=0, related_name='child_groups', null=True, blank=True, to=orm['api_manager.GroupRelationship'])),
('record_active', self.gf('django.db.models.fields.BooleanField')(default=True)),
('record_date_created', self.gf('django.db.models.fields.DateTimeField')(default=datetime.datetime(2014, 3, 27, 0, 0))),
('record_date_modified', self.gf('django.db.models.fields.DateTimeField')(auto_now=True, blank=True)),
))
db.send_create_signal('api_manager', ['GroupRelationship'])
# Adding model 'LinkedGroupRelationship'
db.create_table('api_manager_linkedgrouprelationship', (
('id', self.gf('django.db.models.fields.AutoField')(primary_key=True)),
('from_group_relationship', self.gf('django.db.models.fields.related.ForeignKey')(related_name='from_group_relationships', to=orm['api_manager.GroupRelationship'])),
('to_group_relationship', self.gf('django.db.models.fields.related.ForeignKey')(related_name='to_group_relationships', to=orm['api_manager.GroupRelationship'])),
('record_active', self.gf('django.db.models.fields.BooleanField')(default=True)),
('record_date_created', self.gf('django.db.models.fields.DateTimeField')(default=datetime.datetime(2014, 3, 27, 0, 0))),
('record_date_modified', self.gf('django.db.models.fields.DateTimeField')(auto_now=True, blank=True)),
))
db.send_create_signal('api_manager', ['LinkedGroupRelationship'])
def backwards(self, orm):
# Deleting model 'GroupRelationship'
db.delete_table('api_manager_grouprelationship')
# Deleting model 'LinkedGroupRelationship'
db.delete_table('api_manager_linkedgrouprelationship')
models = {
'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, 3, 27, 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, 3, 27, 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
# pylint: disable=E1101
""" Database ORM models managed by this Django app """
from django.contrib.auth.models import Group
from django.db import models
from django.utils import timezone
class GroupRelationship(models.Model):
"""
The GroupRelationship model contains information describing the relationships of a group,
which allows us to utilize Django's user/group/permission
models and features instead of rolling our own.
"""
group = models.OneToOneField(Group, primary_key=True)
name = models.CharField(max_length=255)
parent_group = models.ForeignKey('self',
related_name="child_groups",
blank=True, null=True, default=0)
linked_groups = models.ManyToManyField('self',
through="LinkedGroupRelationship",
symmetrical=False,
related_name="linked_to+"),
record_active = models.BooleanField(default=True)
record_date_created = models.DateTimeField(default=timezone.now())
record_date_modified = models.DateTimeField(auto_now=True)
def add_linked_group_relationship(self, to_group_relationship, symmetrical=True):
""" Create a new group-group relationship """
relationship = LinkedGroupRelationship.objects.get_or_create(
from_group_relationship=self,
to_group_relationship=to_group_relationship)
if symmetrical:
# avoid recursion by passing `symm=False`
to_group_relationship.add_linked_group_relationship(self, False)
return relationship
def remove_linked_group_relationship(self, to_group_relationship, symmetrical=True):
""" Remove an existing group-group relationship """
LinkedGroupRelationship.objects.filter(
from_group_relationship=self,
to_group_relationship=to_group_relationship).delete()
if symmetrical:
# avoid recursion by passing `symm=False`
to_group_relationship.remove_linked_group_relationship(self, False)
return
def get_linked_group_relationships(self):
""" Retrieve an existing group-group relationship """
efferent_relationships = LinkedGroupRelationship.objects.filter(from_group_relationship=self)
matching_relationships = efferent_relationships
return matching_relationships
def check_linked_group_relationship(self, relationship_to_check, symmetrical=False):
""" Confirm the existence of a possibly-existing group-group relationship """
query = dict(
to_group_relationships__from_group_relationship=self,
to_group_relationships__to_group_relationship=relationship_to_check,
)
if symmetrical:
query.update(
from_group_relationships__to_group_relationship=self,
from_group_relationships__from_group_relationship=relationship_to_check,
)
return GroupRelationship.objects.filter(**query).exists()
class LinkedGroupRelationship(models.Model):
"""
The LinkedGroupRelationship model manages self-referential two-way
relationships between group entities via the GroupRelationship model.
Specifying the intermediary table allows for the definition of additional
relationship information
"""
from_group_relationship = models.ForeignKey(GroupRelationship,
related_name="from_group_relationships",
verbose_name="From Group")
to_group_relationship = models.ForeignKey(GroupRelationship,
related_name="to_group_relationships",
verbose_name="To Group")
record_active = models.BooleanField(default=True)
record_date_created = models.DateTimeField(default=timezone.now())
record_date_modified = models.DateTimeField(auto_now=True)
""" Permissions classes utilized by Django REST Framework """
import logging
from django.conf import settings
from rest_framework import permissions
log = logging.getLogger(__name__)
class ApiKeyHeaderPermission(permissions.BasePermission):
"""
Check for permissions by matching the configured API key and header
"""
def has_permission(self, request, view):
"""
If settings.DEBUG is True and settings.EDX_API_KEY is not set or None,
then allow the request. Otherwise, allow the request if and only if
settings.EDX_API_KEY is set and the X-Edx-Api-Key HTTP header is
present in the request and matches the setting.
"""
debug_enabled = settings.DEBUG
api_key = getattr(settings, "EDX_API_KEY", None)
# DEBUG mode rules over all else
# Including the api_key check here ensures we don't break the feature locally
if debug_enabled and api_key is None:
log.warn("EDX_API_KEY Override: Debug Mode")
return True
# If we're not DEBUG, we need a local api key
if api_key is None:
return False
# The client needs to present the same api key
header_key = request.META['headers'].get('X-Edx-Api-Key')
if header_key is None:
return False
# The api key values need to be the same
if header_key != api_key:
return False
# Allow the request to take place
return True
""" Django REST Framework Serializers """
from django.contrib.auth.models import User
from rest_framework import serializers
class UserSerializer(serializers.ModelSerializer):
""" Serializer for User model interactions """
class Meta:
""" Serializer/field specification """
model = User
fields = ("id", "email", "username")
read_only_fields = ("id", "email", "username")
""" Sessions API URI specification """
from django.conf.urls import patterns, url
urlpatterns = patterns('api_manager.sessions_views',
url(r'/*$^', 'session_list'),
url(r'^(?P<session_id>[a-z0-9]+)$', 'session_detail'),
)
# pylint: disable=E1101
""" API implementation for session-oriented interactions. """
from django.conf import settings
from django.contrib.auth import authenticate, login
from django.contrib.auth import SESSION_KEY, BACKEND_SESSION_KEY, load_backend
from django.contrib.auth.models import AnonymousUser, User
from django.core.exceptions import ObjectDoesNotExist
from django.utils.importlib import import_module
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.serializers import UserSerializer
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
@api_view(['POST'])
@permission_classes((ApiKeyHeaderPermission,))
def session_list(request):
"""
POST creates a new system session, supported authentication modes:
1. Open edX username/password
"""
response_data = {}
base_uri = _generate_base_uri(request)
try:
existing_user = User.objects.get(username=request.DATA['username'])
except ObjectDoesNotExist:
existing_user = None
if existing_user:
user = authenticate(username=existing_user.username, password=request.DATA['password'])
if user is not None:
if user.is_active:
login(request, user)
response_data['token'] = request.session.session_key
response_data['expires'] = request.session.get_expiry_age()
user_dto = UserSerializer(user)
response_data['user'] = user_dto.data
response_data['uri'] = '{}/{}'.format(base_uri, request.session.session_key)
response_status = status.HTTP_201_CREATED
else:
response_status = status.HTTP_403_FORBIDDEN
else:
response_status = status.HTTP_401_UNAUTHORIZED
else:
response_status = status.HTTP_404_NOT_FOUND
return Response(response_data, status=response_status)
@api_view(['GET', 'DELETE'])
@permission_classes((ApiKeyHeaderPermission,))
def session_detail(request, session_id):
"""
GET retrieves an existing system session
DELETE flushes an existing system session from the system
"""
response_data = {}
base_uri = _generate_base_uri(request)
engine = import_module(settings.SESSION_ENGINE)
session = engine.SessionStore(session_id)
if request.method == 'GET':
try:
user_id = session[SESSION_KEY]
backend_path = session[BACKEND_SESSION_KEY]
backend = load_backend(backend_path)
user = backend.get_user(user_id) or AnonymousUser()
except KeyError:
user = AnonymousUser()
if user.is_authenticated():
response_data['token'] = session.session_key
response_data['expires'] = session.get_expiry_age()
response_data['uri'] = base_uri
response_data['user_id'] = user.id
return Response(response_data, status=status.HTTP_200_OK)
else:
return Response(response_data, status=status.HTTP_404_NOT_FOUND)
elif request.method == 'DELETE':
session.flush()
return Response(response_data, status=status.HTTP_204_NO_CONTENT)
""" BASE API VIEWS """
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
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
@api_view(['GET'])
@permission_classes((ApiKeyHeaderPermission,))
def system_detail(request):
"""Returns top-level descriptive information about the Open edX API"""
base_uri = _generate_base_uri(request)
response_data = {}
response_data['name'] = "Open edX System API"
response_data['description'] = "System interface for managing groups, users, and sessions."
response_data['documentation'] = "http://docs.openedxapi.apiary.io/#get-%2Fapi%2Fsystem"
response_data['uri'] = base_uri
return Response(response_data, status=status.HTTP_200_OK)
@api_view(['GET'])
@permission_classes((ApiKeyHeaderPermission,))
def api_detail(request):
"""Returns top-level descriptive information about the Open edX API"""
base_uri = _generate_base_uri(request)
response_data = {}
response_data['name'] = "Open edX API"
response_data['description'] = "Machine interface for interactions with Open edX."
response_data['documentation'] = "http://docs.openedxapi.apiary.io"
response_data['uri'] = base_uri
response_data['resources'] = []
response_data['resources'].append({'uri': base_uri + 'courses'})
response_data['resources'].append({'uri': base_uri + 'groups'})
response_data['resources'].append({'uri': base_uri + 'sessions'})
response_data['resources'].append({'uri': base_uri + 'system'})
response_data['resources'].append({'uri': base_uri + 'users'})
return Response(response_data, status=status.HTTP_200_OK)
# 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)
"""
Run these tests @ Devstack:
rake fasttest_lms[common/djangoapps/api_manager/tests/test_permissions.py]
"""
from random import randint
import unittest
import uuid
from django.conf import settings
from django.test import TestCase
from django.test.utils import override_settings
TEST_API_KEY = "123456ABCDEF"
@override_settings(DEBUG=True, EDX_API_KEY=None)
class PermissionsTestsDebug(TestCase):
""" Test suite for Permissions helper classes """
def setUp(self):
self.test_username = str(uuid.uuid4())
self.test_password = str(uuid.uuid4())
self.test_email = str(uuid.uuid4()) + '@test.org'
def do_post(self, uri, data):
"""Submit an HTTP POST request"""
headers = {
'Content-Type': 'application/json',
'X-Edx-Api-Key': str(TEST_API_KEY),
}
response = self.client.post(uri, headers=headers, data=data)
return response
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_has_permission_debug_enabled(self):
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
local_username = local_username[3:-1] # username is a 32-character field
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 201)
@override_settings(DEBUG=False, EDX_API_KEY="123456ABCDEF")
class PermissionsTestsApiKey(TestCase):
""" Test suite for Permissions helper classes """
def setUp(self):
self.test_username = str(uuid.uuid4())
self.test_password = str(uuid.uuid4())
self.test_email = str(uuid.uuid4()) + '@test.org'
def do_post(self, uri, data):
"""Submit an HTTP POST request"""
headers = {
'Content-Type': 'application/json',
'X-Edx-Api-Key': str(TEST_API_KEY),
}
response = self.client.post(uri, headers=headers, data=data)
return response
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_has_permission_valid_api_key(self):
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
local_username = local_username[3:-1] # username is a 32-character field
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 201)
@override_settings(DEBUG=False, EDX_API_KEY=None)
class PermissionsTestDeniedMissingServerKey(TestCase):
""" Test suite for Permissions helper classes """
def setUp(self):
self.test_username = str(uuid.uuid4())
self.test_password = str(uuid.uuid4())
self.test_email = str(uuid.uuid4()) + '@test.org'
def do_post(self, uri, data):
"""Submit an HTTP POST request"""
headers = {
'Content-Type': 'application/json',
'X-Edx-Api-Key': str(TEST_API_KEY),
}
response = self.client.post(uri, headers=headers, data=data)
return response
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_has_permission_missing_server_key(self):
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
local_username = local_username[3:-1] # username is a 32-character field
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 403)
@override_settings(DEBUG=False, EDX_API_KEY="67890VWXYZ")
class PermissionsTestDeniedMissingClientKey(TestCase):
""" Test suite for Permissions helper classes """
def setUp(self):
self.test_username = str(uuid.uuid4())
self.test_password = str(uuid.uuid4())
self.test_email = str(uuid.uuid4()) + '@test.org'
def do_post(self, uri, data):
"""Submit an HTTP POST request"""
headers = {
'Content-Type': 'application/json',
}
response = self.client.post(uri, headers=headers, data=data)
return response
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_has_permission_invalid_client_key(self):
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
local_username = local_username[3:-1] # username is a 32-character field
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 403)
@override_settings(DEBUG=False, EDX_API_KEY="67890VWXYZ")
class PermissionsTestDeniedInvalidClientKey(TestCase):
""" Test suite for Permissions helper classes """
def setUp(self):
self.test_username = str(uuid.uuid4())
self.test_password = str(uuid.uuid4())
self.test_email = str(uuid.uuid4()) + '@test.org'
def do_post(self, uri, data):
"""Submit an HTTP POST request"""
headers = {
'Content-Type': 'application/json',
'X-Edx-Api-Key': str(TEST_API_KEY),
}
response = self.client.post(uri, headers=headers, data=data)
return response
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_has_permission_invalid_client_key(self):
test_uri = '/api/users'
local_username = self.test_username + str(randint(11, 99))
local_username = local_username[3:-1] # username is a 32-character field
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password}
response = self.do_post(test_uri, data)
self.assertEqual(response.status_code, 403)
# pylint: disable=E1101
# pylint: disable=E1103
"""
Run these tests @ Devstack:
rake fasttest_lms[common/djangoapps/api_manager/tests/test_session_views.py]
"""
from random import randint
import unittest
import uuid
from django.conf import settings
from django.contrib.auth.models import User
from django.core.cache import cache
from django.test import TestCase, Client
from django.test.utils import override_settings
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(EDX_API_KEY=TEST_API_KEY)
class SessionsApiTests(TestCase):
""" Test suite for Sessions API views """
def setUp(self):
self.test_server_prefix = 'https://testserver'
self.test_username = str(uuid.uuid4())
self.test_password = str(uuid.uuid4())
self.test_email = str(uuid.uuid4()) + '@test.org'
self.base_users_uri = '/api/users'
self.base_sessions_uri = '/api/sessions'
self.client = SecureClient()
cache.clear()
def do_post(self, uri, data):
"""Submit an HTTP POST request"""
headers = {
'Content-Type': 'application/json',
'X-Edx-Api-Key': str(TEST_API_KEY),
}
response = self.client.post(uri, headers=headers, data=data)
return response
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_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
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_session_list_post_valid(self):
local_username = self.test_username + str(randint(11, 99))
local_username = local_username[3:-1] # username is a 32-character field
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password}
response = self.do_post(self.base_users_uri, data)
user_id = response.data['id']
data = {'username': local_username, 'password': self.test_password}
response = self.do_post(self.base_sessions_uri, data)
self.assertEqual(response.status_code, 201)
self.assertGreater(len(response.data['token']), 0)
confirm_uri = self.test_server_prefix + self.base_sessions_uri + '/' + response.data['token']
self.assertEqual(response.data['uri'], confirm_uri)
self.assertGreater(response.data['expires'], 0)
self.assertGreater(len(response.data['user']), 0)
self.assertEqual(str(response.data['user']['username']), local_username)
self.assertEqual(response.data['user']['id'], user_id)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_session_list_post_invalid(self):
local_username = self.test_username + str(randint(11, 99))
local_username = local_username[3:-1] # username is a 32-character field
bad_password = "12345"
data = {'email': self.test_email, 'username': local_username, 'password': bad_password}
response = self.do_post(self.base_users_uri, data)
data = {'username': local_username, 'password': self.test_password}
response = self.do_post(self.base_sessions_uri, data)
self.assertEqual(response.status_code, 401)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_session_list_post_valid_inactive(self):
local_username = self.test_username + str(randint(11, 99))
local_username = local_username[3:-1] # username is a 32-character field
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password}
response = self.do_post(self.base_users_uri, data)
user = User.objects.get(username=local_username)
user.is_active = False
user.save()
data = {'username': local_username, 'password': self.test_password}
response = self.do_post(self.base_sessions_uri, data)
self.assertEqual(response.status_code, 403)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_session_list_post_invalid_notfound(self):
data = {'username': 'user_12321452334', 'password': self.test_password}
response = self.do_post(self.base_sessions_uri, data)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_session_detail_get(self):
local_username = self.test_username + str(randint(11, 99))
local_username = local_username[3:-1] # username is a 32-character field
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password}
response = self.do_post(self.base_users_uri, data)
data = {'username': local_username, 'password': self.test_password}
response = self.do_post(self.base_sessions_uri, data)
test_uri = self.base_sessions_uri + '/' + response.data['token']
post_token = response.data['token']
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data['token'], post_token)
response = self.do_delete(test_uri)
self.assertEqual(response.status_code, 204)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_session_detail_get_undefined(self):
test_uri = self.base_sessions_uri + "/123456789"
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_session_detail_delete(self):
local_username = self.test_username + str(randint(11, 99))
local_username = local_username[3:-1] # username is a 32-character field
data = {'email': self.test_email, 'username': local_username, 'password': self.test_password}
response = self.do_post(self.base_users_uri, data)
self.assertEqual(response.status_code, 201)
data = {'username': local_username, 'password': self.test_password}
response = self.do_post(self.base_sessions_uri, data)
self.assertEqual(response.status_code, 201)
test_uri = self.base_users_uri + str(response.data['user']['id'])
response = self.do_delete(test_uri)
self.assertEqual(response.status_code, 204)
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 404)
# pylint: disable=E1103
"""
Run these tests @ Devstack:
rake fasttest_lms[common/djangoapps/api_manager/tests/test_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
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(EDX_API_KEY=TEST_API_KEY)
class SystemApiTests(TestCase):
""" Test suite for base API views """
def setUp(self):
self.test_server_prefix = "https://testserver/api"
self.test_username = str(uuid.uuid4())
self.test_password = str(uuid.uuid4())
self.test_email = str(uuid.uuid4()) + '@test.org'
self.test_group_name = str(uuid.uuid4())
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
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_system_detail_get(self):
""" Ensure the system returns base data about the system """
test_uri = self.test_server_prefix + '/system'
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertIsNotNone(response.data['uri'])
self.assertGreater(len(response.data['uri']), 0)
self.assertEqual(response.data['uri'], test_uri)
self.assertIsNotNone(response.data['documentation'])
self.assertGreater(len(response.data['documentation']), 0)
self.assertIsNotNone(response.data['name'])
self.assertGreater(len(response.data['name']), 0)
self.assertIsNotNone(response.data['description'])
self.assertGreater(len(response.data['description']), 0)
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
def test_system_detail_api_get(self):
""" Ensure the system returns base data about the API """
test_uri = self.test_server_prefix
response = self.do_get(test_uri)
self.assertEqual(response.status_code, 200)
self.assertIsNotNone(response.data['uri'])
self.assertGreater(len(response.data['uri']), 0)
self.assertEqual(response.data['uri'], test_uri)
self.assertIsNotNone(response.data['documentation'])
self.assertGreater(len(response.data['documentation']), 0)
self.assertIsNotNone(response.data['name'])
self.assertGreater(len(response.data['name']), 0)
self.assertIsNotNone(response.data['description'])
self.assertGreater(len(response.data['description']), 0)
"""
The URI scheme for resources is as follows:
Resource type: /api/{resource_type}
Specific resource: /api/{resource_type}/{resource_id}
The remaining URIs provide information about the API and/or module
System: General context and intended usage
API: Top-level description of overall API (must live somewhere)
"""
from django.conf.urls import include, patterns, url
urlpatterns = patterns('api_manager.system_views',
url(r'^$', 'api_detail'),
url(r'^system$', 'system_detail'),
url(r'^users/*', include('api_manager.users_urls')),
url(r'^groups/*', include('api_manager.groups_urls')),
url(r'^sessions/*', include('api_manager.sessions_urls')),
url(r'^courses/*', include('api_manager.courses_urls')),
)
""" Users API URI specification """
from django.conf.urls import patterns, url
urlpatterns = patterns('api_manager.users_views',
url(r'/*$^', 'user_list'),
url(r'^(?P<user_id>[0-9]+)$', 'user_detail'),
url(r'^(?P<user_id>[0-9]+)/courses/*$', 'user_courses_list'),
url(r'^(?P<user_id>[0-9]+)/courses/(?P<course_id>[a-zA-Z0-9/_:]+)$', 'user_courses_detail'),
url(r'^(?P<user_id>[0-9]+)/groups/*$', 'user_groups_list'),
url(r'^(?P<user_id>[0-9]+)/groups/(?P<group_id>[0-9]+)$', 'user_groups_detail'),
)
...@@ -169,6 +169,7 @@ def save_child_position(seq_module, child_name): ...@@ -169,6 +169,7 @@ def save_child_position(seq_module, child_name):
""" """
child_name: url_name of the child child_name: url_name of the child
""" """
print child_name
for position, c in enumerate(seq_module.get_display_items(), start=1): for position, c in enumerate(seq_module.get_display_items(), start=1):
if c.url_name == child_name: if c.url_name == child_name:
# Only save if position changed # Only save if position changed
......
...@@ -249,6 +249,9 @@ FEATURES = { ...@@ -249,6 +249,9 @@ FEATURES = {
# Turn off Advanced Security by default # Turn off Advanced Security by default
'ADVANCED_SECURITY': False, 'ADVANCED_SECURITY': False,
# Turn on/off the Open edX API
'API': False,
} }
# Used for A/B testing # Used for A/B testing
...@@ -1203,6 +1206,9 @@ INSTALLED_APPS = ( ...@@ -1203,6 +1206,9 @@ INSTALLED_APPS = (
# Monitoring functionality # Monitoring functionality
'monitoring', 'monitoring',
# EDX API application
'api_manager',
) )
######################### MARKETING SITE ############################### ######################### MARKETING SITE ###############################
......
...@@ -26,7 +26,6 @@ EMAIL_BACKEND = 'django.core.mail.backends.console.EmailBackend' ...@@ -26,7 +26,6 @@ EMAIL_BACKEND = 'django.core.mail.backends.console.EmailBackend'
FEATURES['ENABLE_INSTRUCTOR_EMAIL'] = True # Enable email for all Studio courses FEATURES['ENABLE_INSTRUCTOR_EMAIL'] = True # Enable email for all Studio courses
FEATURES['REQUIRE_COURSE_EMAIL_AUTH'] = False # Give all courses email (don't require django-admin perms) FEATURES['REQUIRE_COURSE_EMAIL_AUTH'] = False # Give all courses email (don't require django-admin perms)
################################ DEBUG TOOLBAR ################################ ################################ DEBUG TOOLBAR ################################
INSTALLED_APPS += ('debug_toolbar',) INSTALLED_APPS += ('debug_toolbar',)
...@@ -66,6 +65,11 @@ FEATURES['AUTOMATIC_VERIFY_STUDENT_IDENTITY_FOR_TESTING'] = True ...@@ -66,6 +65,11 @@ FEATURES['AUTOMATIC_VERIFY_STUDENT_IDENTITY_FOR_TESTING'] = True
FEATURES['ENABLE_PAYMENT_FAKE'] = True FEATURES['ENABLE_PAYMENT_FAKE'] = True
CC_PROCESSOR['CyberSource']['PURCHASE_ENDPOINT'] = '/shoppingcart/payment_fake/' CC_PROCESSOR['CyberSource']['PURCHASE_ENDPOINT'] = '/shoppingcart/payment_fake/'
########################### EDX API #################################
FEATURES['API'] = True
##################################################################### #####################################################################
# Lastly, see if the developer has any local overrides. # Lastly, see if the developer has any local overrides.
try: try:
......
...@@ -44,6 +44,9 @@ FEATURES['ALLOW_COURSE_STAFF_GRADE_DOWNLOADS'] = True ...@@ -44,6 +44,9 @@ FEATURES['ALLOW_COURSE_STAFF_GRADE_DOWNLOADS'] = True
# Toggles embargo on for testing # Toggles embargo on for testing
FEATURES['EMBARGO'] = True FEATURES['EMBARGO'] = True
# Toggles API on for testing
FEATURES['API'] = True
# Need wiki for courseware views to work. TODO (vshnayder): shouldn't need it. # Need wiki for courseware views to work. TODO (vshnayder): shouldn't need it.
WIKI_ENABLED = True WIKI_ENABLED = True
......
...@@ -67,8 +67,15 @@ urlpatterns = ('', # nopep8 ...@@ -67,8 +67,15 @@ urlpatterns = ('', # nopep8
url(r'^i18n/', include('django.conf.urls.i18n')), url(r'^i18n/', include('django.conf.urls.i18n')),
url(r'^embargo$', 'student.views.embargo', name="embargo"), url(r'^embargo$', 'student.views.embargo', name="embargo"),
) )
# OPEN EDX API
if settings.FEATURES["API"]:
urlpatterns += (
url(r'^api/*', include('api_manager.urls')),
)
# if settings.FEATURES.get("MULTIPLE_ENROLLMENT_ROLES"): # if settings.FEATURES.get("MULTIPLE_ENROLLMENT_ROLES"):
urlpatterns += ( urlpatterns += (
url(r'^verify_student/', include('verify_student.urls')), url(r'^verify_student/', include('verify_student.urls')),
......
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