Commit e63194c1 by Giovanni Di Milia

Added CCX REST APIs

CCX REST APIs
OAUTH2 authorization for CCX APIs
- oauth2 authorization required for ccx list.
- Course-instructor permission for ccx api endpoint
- Protection for detail view too.

Tests for CCX REST APIs and OAUTH2 authorization
parent f59144b7
""" CCX API URLs. """
from django.conf.urls import patterns, url, include
urlpatterns = patterns(
'',
url(r'^v0/', include('lms.djangoapps.ccx.api.v0.urls', namespace='v0')),
)
""" CCX API v0 Paginators. """
from openedx.core.lib.api.paginators import DefaultPagination
class CCXAPIPagination(DefaultPagination):
"""
Pagination format used by the CCX API.
"""
page_size_query_param = "page_size"
def get_paginated_response(self, data):
"""
Annotate the response with pagination information.
"""
response = super(CCXAPIPagination, self).get_paginated_response(data)
# Add the current page to the response.
response.data["current_page"] = self.page.number
# This field can be derived from other fields in the response,
# so it may make sense to have the JavaScript client calculate it
# instead of including it in the response.
response.data["start"] = (self.page.number - 1) * self.get_page_size(self.request)
return response
""" CCX API v0 Serializers. """
from rest_framework import serializers
from lms.djangoapps.ccx.models import CustomCourseForEdX
from ccx_keys.locator import CCXLocator
class CCXCourseSerializer(serializers.ModelSerializer):
"""
Serializer for CCX courses
"""
ccx_course_id = serializers.SerializerMethodField()
master_course_id = serializers.CharField(source='course_id')
display_name = serializers.CharField()
coach_email = serializers.EmailField(source='coach.email')
start = serializers.CharField(allow_blank=True)
due = serializers.CharField(allow_blank=True)
max_students_allowed = serializers.IntegerField(source='max_student_enrollments_allowed')
class Meta(object):
model = CustomCourseForEdX
fields = (
"ccx_course_id",
"master_course_id",
"display_name",
"coach_email",
"start",
"due",
"max_students_allowed",
)
read_only_fields = (
"ccx_course_id",
"master_course_id",
"start",
"due",
)
@staticmethod
def get_ccx_course_id(obj):
"""
Getter for the CCX Course ID
"""
return unicode(CCXLocator.from_course_locator(obj.course.id, obj.id))
"""
Tests for the CCX REST APIs.
"""
import datetime
import json
import math
import pytz
import string
import urllib
import urlparse
import ddt
import mock
from django.conf import settings
from django.contrib.auth.models import User
from django.core.urlresolvers import (
reverse,
resolve,
Resolver404
)
from nose.plugins.attrib import attr
from provider.constants import CONFIDENTIAL
from provider.oauth2.models import (
Client,
Grant,
)
from rest_framework import status
from rest_framework.test import APITestCase
from courseware import courses
from ccx_keys.locator import CCXLocator
from student.models import CourseEnrollment
from instructor.enrollment import (
enroll_email,
get_email_params,
)
from lms.djangoapps.ccx.api.v0 import views
from lms.djangoapps.ccx.models import CcxFieldOverride, CustomCourseForEdX
from lms.djangoapps.ccx.overrides import override_field_for_ccx
from lms.djangoapps.ccx.tests.utils import CcxTestCase
from opaque_keys.edx.keys import CourseKey
from student.roles import CourseCcxCoachRole
from student.tests.factories import AdminFactory
class CcxRestApiTest(CcxTestCase, APITestCase):
"""
Base class with common methods to be used in the test classes of this module
"""
@classmethod
def setUpClass(cls):
super(CcxRestApiTest, cls).setUpClass()
def setUp(self):
"""
Set up tests
"""
super(CcxRestApiTest, self).setUp()
# add some info about the course for easy access
self.master_course_key = self.course.location.course_key
self.master_course_key_str = unicode(self.master_course_key)
# OAUTH2 setup
# create a specific user for the application
app_user = User.objects.create_user('test_app_user', 'test_app_user@openedx.org', 'test')
# create an oauth client app entry
self.app_client = Client.objects.create(
user=app_user,
name='test client',
url='http://localhost//',
redirect_uri='http://localhost//',
client_type=CONFIDENTIAL
)
# create an authorization code
self.app_grant = Grant.objects.create(
user=app_user,
client=self.app_client,
redirect_uri='http://localhost//'
)
self.course.enable_ccx = True
self.mstore.update_item(self.course, self.coach.id)
self.auth = self.get_auth_token()
def get_auth_token(self):
"""
Helper method to get the oauth token
"""
token_data = {
'grant_type': 'authorization_code',
'code': self.app_grant.code,
'client_id': self.app_client.client_id,
'client_secret': self.app_client.client_secret
}
token_resp = self.client.post('/oauth2/access_token/', data=token_data)
self.assertEqual(token_resp.status_code, status.HTTP_200_OK)
token_resp_json = json.loads(token_resp.content)
self.assertIn('access_token', token_resp_json)
return 'Bearer {0}'.format(token_resp_json.get('access_token'))
def expect_error(self, http_code, error_code_str, resp_obj):
"""
Helper function that checks that the response object
has a body with the provided error
"""
self.assertEqual(resp_obj.status_code, http_code)
self.assertIn('error_code', resp_obj.data)
self.assertEqual(resp_obj.data['error_code'], error_code_str)
def expect_error_fields(self, expected_field_errors, resp_obj):
"""
Helper function that checks that the response object
has a body with the provided field errors
"""
self.assertEqual(resp_obj.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn('field_errors', resp_obj.data)
# restructure the error dictionary for a easier comparison
resp_dict_error = {}
for field_name, error_dict in resp_obj.data['field_errors'].iteritems():
resp_dict_error[field_name] = error_dict.get('error_code', '')
self.assertEqual(expected_field_errors, resp_dict_error)
@attr('shard_1')
@ddt.ddt
class CcxListTest(CcxRestApiTest):
"""
Test for the CCX REST APIs
"""
@classmethod
def setUpClass(cls):
super(CcxListTest, cls).setUpClass()
def setUp(self):
"""
Set up tests
"""
super(CcxListTest, self).setUp()
self.list_url = reverse('ccx_api:v0:ccx:list')
def test_authorization(self):
"""
Test that only the right token is authorized
"""
url = urlparse.urljoin(
self.list_url,
'?master_course_id={0}'.format(urllib.quote_plus(self.master_course_key_str))
)
auth_list = [
"Wrong token-type-obviously",
"Bearer wrong token format",
"Bearer wrong-token",
"Bearer",
"Bearer hfbhfbfwq398248fnid939rh3489fh39nd4m34r9" # made up token
]
# all the auths in the list fail to authorize
for auth in auth_list:
resp = self.client.get(url, {}, HTTP_AUTHORIZATION=auth)
self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED)
resp = self.client.get(url, {}, HTTP_AUTHORIZATION=self.auth)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
def test_get_list_wrong_master_course(self):
"""
Test for various get requests with wrong master course string
"""
# case with no master_course_id provided
resp = self.client.get(self.list_url, {}, HTTP_AUTHORIZATION=self.auth)
self.expect_error(status.HTTP_400_BAD_REQUEST, 'master_course_id_not_provided', resp)
base_url = urlparse.urljoin(self.list_url, '?master_course_id=')
# case with empty master_course_id
resp = self.client.get(base_url, {}, HTTP_AUTHORIZATION=self.auth)
self.expect_error(status.HTTP_400_BAD_REQUEST, 'course_id_not_valid', resp)
# case with invalid master_course_id
url = '{0}invalid_master_course_str'.format(base_url)
resp = self.client.get(url, {}, HTTP_AUTHORIZATION=self.auth)
self.expect_error(status.HTTP_400_BAD_REQUEST, 'course_id_not_valid', resp)
# case with inexistent master_course_id
url = '{0}course-v1%3Aorg_foo.0%2Bcourse_bar_0%2BRun_0'.format(base_url)
resp = self.client.get(url, {}, HTTP_AUTHORIZATION=self.auth)
self.expect_error(status.HTTP_404_NOT_FOUND, 'course_id_does_not_exist', resp)
def test_get_list(self):
"""
Tests the API to get a list of CCX Courses
"""
# get the list of ccx
url = urlparse.urljoin(
self.list_url,
'?master_course_id={0}'.format(urllib.quote_plus(self.master_course_key_str))
)
# there are no CCX courses
resp = self.client.get(url, {}, HTTP_AUTHORIZATION=self.auth)
self.assertIn('count', resp.data) # pylint: disable=no-member
self.assertEqual(resp.data['count'], 0) # pylint: disable=no-member
# create few ccx courses
num_ccx = 10
for _ in xrange(num_ccx):
self.make_ccx()
resp = self.client.get(url, {}, HTTP_AUTHORIZATION=self.auth)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
self.assertIn('count', resp.data) # pylint: disable=no-member
self.assertEqual(resp.data['count'], num_ccx) # pylint: disable=no-member
self.assertIn('results', resp.data) # pylint: disable=no-member
self.assertEqual(len(resp.data['results']), num_ccx) # pylint: disable=no-member
def test_get_sorted_list(self):
"""
Tests the API to get a sorted list of CCX Courses
"""
# create few ccx courses
num_ccx = 3
for _ in xrange(num_ccx):
self.make_ccx()
# update the display_name fields
all_ccx = CustomCourseForEdX.objects.all()
all_ccx = all_ccx.order_by('id')
self.assertEqual(len(all_ccx), num_ccx)
title_str = 'Title CCX {0}'
for num, ccx in enumerate(all_ccx):
ccx.display_name = title_str.format(string.ascii_lowercase[-(num + 1)])
ccx.save()
# get the list of ccx
base_url = urlparse.urljoin(
self.list_url,
'?master_course_id={0}'.format(urllib.quote_plus(self.master_course_key_str))
)
# sort by display name
url = '{0}&order_by=display_name'.format(base_url)
resp = self.client.get(url, {}, HTTP_AUTHORIZATION=self.auth)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
self.assertEqual(len(resp.data['results']), num_ccx) # pylint: disable=no-member
# the display_name should be sorted as "Title CCX x", "Title CCX y", "Title CCX z"
for num, ccx in enumerate(resp.data['results']): # pylint: disable=no-member
self.assertEqual(title_str.format(string.ascii_lowercase[-(num_ccx - num)]), ccx['display_name'])
# add sort order desc
url = '{0}&order_by=display_name&sort_order=desc'.format(base_url)
resp = self.client.get(url, {}, HTTP_AUTHORIZATION=self.auth)
# the only thing I can check is that the display name is in alphabetically reversed order
# in the same way when the field has been updated above, so with the id asc
for num, ccx in enumerate(resp.data['results']): # pylint: disable=no-member
self.assertEqual(title_str.format(string.ascii_lowercase[-(num + 1)]), ccx['display_name'])
def test_get_paginated_list(self):
"""
Tests the API to get a paginated list of CCX Courses
"""
# create some ccx courses
num_ccx = 357
for _ in xrange(num_ccx):
self.make_ccx()
# get the list of ccx
base_url = urlparse.urljoin(
self.list_url,
'?master_course_id={0}'.format(urllib.quote_plus(self.master_course_key_str))
)
page_size = settings.REST_FRAMEWORK.get('PAGE_SIZE', 10)
num_pages = int(math.ceil(num_ccx / float(page_size)))
# get first page
resp = self.client.get(base_url, {}, HTTP_AUTHORIZATION=self.auth)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
self.assertEqual(resp.data['count'], num_ccx) # pylint: disable=no-member
self.assertEqual(resp.data['num_pages'], num_pages) # pylint: disable=no-member
self.assertEqual(resp.data['current_page'], 1) # pylint: disable=no-member
self.assertEqual(resp.data['start'], 0) # pylint: disable=no-member
self.assertIsNotNone(resp.data['next']) # pylint: disable=no-member
self.assertIsNone(resp.data['previous']) # pylint: disable=no-member
# get a page in the middle
url = '{0}&page=24'.format(base_url)
resp = self.client.get(url, {}, HTTP_AUTHORIZATION=self.auth)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
self.assertEqual(resp.data['count'], num_ccx) # pylint: disable=no-member
self.assertEqual(resp.data['num_pages'], num_pages) # pylint: disable=no-member
self.assertEqual(resp.data['current_page'], 24) # pylint: disable=no-member
self.assertEqual(resp.data['start'], (resp.data['current_page'] - 1) * page_size) # pylint: disable=no-member
self.assertIsNotNone(resp.data['next']) # pylint: disable=no-member
self.assertIsNotNone(resp.data['previous']) # pylint: disable=no-member
# get last page
url = '{0}&page={1}'.format(base_url, num_pages)
resp = self.client.get(url, {}, HTTP_AUTHORIZATION=self.auth)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
self.assertEqual(resp.data['count'], num_ccx) # pylint: disable=no-member
self.assertEqual(resp.data['num_pages'], num_pages) # pylint: disable=no-member
self.assertEqual(resp.data['current_page'], num_pages) # pylint: disable=no-member
self.assertEqual(resp.data['start'], (resp.data['current_page'] - 1) * page_size) # pylint: disable=no-member
self.assertIsNone(resp.data['next']) # pylint: disable=no-member
self.assertIsNotNone(resp.data['previous']) # pylint: disable=no-member
# last page + 1
url = '{0}&page={1}'.format(base_url, num_pages + 1)
resp = self.client.get(url, {}, HTTP_AUTHORIZATION=self.auth)
self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
@ddt.data(
(
{},
status.HTTP_400_BAD_REQUEST,
'master_course_id_not_provided'
),
(
{'master_course_id': None},
status.HTTP_400_BAD_REQUEST,
'master_course_id_not_provided'
),
(
{'master_course_id': ''},
status.HTTP_400_BAD_REQUEST,
'course_id_not_valid'
),
(
{'master_course_id': 'invalid_master_course_str'},
status.HTTP_400_BAD_REQUEST,
'course_id_not_valid'
),
(
{'master_course_id': 'course-v1:org_foo.0+course_bar_0+Run_0'},
status.HTTP_404_NOT_FOUND,
'course_id_does_not_exist'
),
)
@ddt.unpack
def test_post_list_wrong_master_course(self, data, expected_http_error, expected_error_string):
"""
Test for various post requests with wrong master course string
"""
# case with no master_course_id provided
resp = self.client.post(self.list_url, data, format='json', HTTP_AUTHORIZATION=self.auth)
self.expect_error(expected_http_error, expected_error_string, resp)
def test_post_list_wrong_master_course_special_cases(self):
"""
Same as test_post_list_wrong_master_course,
but different ways to test the wrong master_course_id
"""
# case with ccx not enabled for master_course_id
self.course.enable_ccx = False
self.mstore.update_item(self.course, self.coach.id)
data = {'master_course_id': self.master_course_key_str}
resp = self.client.post(self.list_url, data, format='json', HTTP_AUTHORIZATION=self.auth)
self.expect_error(status.HTTP_403_FORBIDDEN, 'ccx_not_enabled_for_master_course', resp)
self.course.enable_ccx = True
self.mstore.update_item(self.course, self.coach.id)
# case with deprecated master_course_id
with mock.patch('courseware.courses.get_course_by_id', autospec=True) as mocked:
mocked.return_value.id.deprecated = True
resp = self.client.post(self.list_url, data, format='json', HTTP_AUTHORIZATION=self.auth)
self.expect_error(status.HTTP_400_BAD_REQUEST, 'deprecated_master_course_id', resp)
@ddt.data(
(
{},
{
'max_students_allowed': 'missing_field_max_students_allowed',
'display_name': 'missing_field_display_name',
'coach_email': 'missing_field_coach_email'
}
),
(
{
'max_students_allowed': 10,
'display_name': 'CCX Title'
},
{
'coach_email': 'missing_field_coach_email'
}
),
(
{
'max_students_allowed': None,
'display_name': None,
'coach_email': None
},
{
'max_students_allowed': 'null_field_max_students_allowed',
'display_name': 'null_field_display_name',
'coach_email': 'null_field_coach_email'
}
),
(
{
'max_students_allowed': 10,
'display_name': 'CCX Title',
'coach_email': 'this is not an email@test.com'
},
{'coach_email': 'invalid_coach_email'}
),
(
{
'max_students_allowed': 10,
'display_name': '',
'coach_email': 'email@test.com'
},
{'display_name': 'invalid_display_name'}
),
(
{
'max_students_allowed': 'a',
'display_name': 'CCX Title',
'coach_email': 'email@test.com'
},
{'max_students_allowed': 'invalid_max_students_allowed'}
),
)
@ddt.unpack
def test_post_list_wrong_input_data(self, data, expected_errors):
"""
Test for various post requests with wrong master course string
"""
# add the master_course_key_str to the request data
data['master_course_id'] = self.master_course_key_str
resp = self.client.post(self.list_url, data, format='json', HTTP_AUTHORIZATION=self.auth)
self.expect_error_fields(expected_errors, resp)
def test_post_list_coach_does_not_exist(self):
"""
Specific test for the case when the input data is valid but the coach does not exist.
"""
data = {
'master_course_id': self.master_course_key_str,
'max_students_allowed': 111,
'display_name': 'CCX Title',
'coach_email': 'inexisting_email@test.com'
}
resp = self.client.post(self.list_url, data, format='json', HTTP_AUTHORIZATION=self.auth)
self.expect_error(status.HTTP_404_NOT_FOUND, 'coach_user_does_not_exist', resp)
def test_post_list(self):
"""
Test the creation of a CCX
"""
outbox = self.get_outbox()
data = {
'master_course_id': self.master_course_key_str,
'max_students_allowed': 111,
'display_name': 'CCX Test Title',
'coach_email': self.coach.email
}
resp = self.client.post(self.list_url, data, format='json', HTTP_AUTHORIZATION=self.auth)
self.assertEqual(resp.status_code, status.HTTP_201_CREATED)
# check if the response has at least the same data of the request
for key, val in data.iteritems():
self.assertEqual(resp.data.get(key), val) # pylint: disable=no-member
self.assertIn('ccx_course_id', resp.data) # pylint: disable=no-member
# check that the new CCX actually exists
course_key = CourseKey.from_string(resp.data.get('ccx_course_id')) # pylint: disable=no-member
ccx_course = CustomCourseForEdX.objects.get(pk=course_key.ccx)
self.assertEqual(
unicode(CCXLocator.from_course_locator(ccx_course.course.id, ccx_course.id)),
resp.data.get('ccx_course_id') # pylint: disable=no-member
)
# check that the coach user has coach role on the master course
coach_role_on_master_course = CourseCcxCoachRole(self.master_course_key)
self.assertTrue(coach_role_on_master_course.has_user(self.coach))
# check that the coach has been enrolled in the ccx
ccx_course_object = courses.get_course_by_id(course_key)
self.assertTrue(
CourseEnrollment.objects.filter(course_id=ccx_course_object.id, user=self.coach).exists()
)
# check that an email has been sent to the coach
self.assertEqual(len(outbox), 1)
self.assertIn(self.coach.email, outbox[0].recipients()) # pylint: disable=no-member
@attr('shard_1')
@ddt.ddt
class CcxDetailTest(CcxRestApiTest):
"""
Test for the CCX REST APIs
"""
@classmethod
def setUpClass(cls):
super(CcxDetailTest, cls).setUpClass()
def setUp(self):
"""
Set up tests
"""
super(CcxDetailTest, self).setUp()
self.make_coach()
# create a ccx
self.ccx = self.make_ccx(max_students_allowed=123)
self.ccx_key = CCXLocator.from_course_locator(self.ccx.course.id, self.ccx.id)
self.ccx_key_str = unicode(self.ccx_key)
self.detail_url = reverse('ccx_api:v0:ccx:detail', kwargs={'ccx_course_id': self.ccx_key_str})
def make_ccx(self, max_students_allowed=200):
"""
Overridden method to replicate (part of) the actual
creation of ccx courses
"""
ccx = super(CcxDetailTest, self).make_ccx(max_students_allowed=max_students_allowed)
today = datetime.datetime.today()
start = today.replace(tzinfo=pytz.UTC)
override_field_for_ccx(ccx, self.course, 'start', start)
override_field_for_ccx(ccx, self.course, 'due', None)
# Hide anything that can show up in the schedule
hidden = 'visible_to_staff_only'
for chapter in self.course.get_children():
override_field_for_ccx(ccx, chapter, hidden, True)
for sequential in chapter.get_children():
override_field_for_ccx(ccx, sequential, hidden, True)
for vertical in sequential.get_children():
override_field_for_ccx(ccx, vertical, hidden, True)
# enroll the coach in the CCX
ccx_course_key = CCXLocator.from_course_locator(self.course.id, ccx.id)
email_params = get_email_params(
self.course,
auto_enroll=True,
course_key=ccx_course_key,
display_name=ccx.display_name
)
enroll_email(
course_id=ccx_course_key,
student_email=self.coach.email,
auto_enroll=True,
email_students=False,
email_params=email_params,
)
return ccx
def test_authorization(self):
"""
Test that only the right token is authorized
"""
auth_list = [
"Wrong token-type-obviously",
"Bearer wrong token format",
"Bearer wrong-token",
"Bearer",
"Bearer hfbhfbfwq398248fnid939rh3489fh39nd4m34r9" # made up token
]
# all the auths in the list fail to authorize
for auth in auth_list:
resp = self.client.get(self.detail_url, {}, HTTP_AUTHORIZATION=auth)
self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED)
resp = self.client.get(self.detail_url, {}, HTTP_AUTHORIZATION=self.auth)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
def test_resolve_get_detail(self):
"""
Test for the ccx detail view resolver. This is needed because it is assumed
that only an URL with a valid course id string can reach the detail view.
"""
# get the base url from the valid one to build invalid urls
base_url = '{0}/'.format(self.detail_url.rsplit('/', 1)[0])
# this url should be the same of the ccx list view
resolver = resolve(base_url)
self.assertEqual(views.CCXListView.__name__, resolver.func.__name__)
self.assertEqual(views.CCXListView.__module__, resolver.func.__module__)
# invalid urls
for invalid_ccx_id in ('foo', 'ccx-v1:org.0', 'ccx-v1:org.0+course_0'):
with self.assertRaises(Resolver404):
resolve('{0}{1}'.format(base_url, invalid_ccx_id))
# the following course ID works even if it is not a CCX valid course id (the regex matches course ID strings)
resolver = resolve('{0}{1}'.format(base_url, 'ccx-v1:org.0+course_0+Run_0'))
self.assertEqual(views.CCXDetailView.__name__, resolver.func.__name__)
self.assertEqual(views.CCXDetailView.__module__, resolver.func.__module__)
# and of course a valid ccx course id
resolver = resolve('{0}{1}'.format(base_url, self.ccx_key_str))
self.assertEqual(views.CCXDetailView.__name__, resolver.func.__name__)
self.assertEqual(views.CCXDetailView.__module__, resolver.func.__module__)
@ddt.data(('get',), ('delete',), ('patch',))
@ddt.unpack
def test_detail_wrong_ccx(self, http_method):
"""
Test for different methods for detail of a ccx course.
All check the validity of the ccx course id
"""
client_request = getattr(self.client, http_method)
# get a detail url with a master_course id string
url = reverse('ccx_api:v0:ccx:detail', kwargs={'ccx_course_id': self.master_course_key_str})
resp = client_request(url, {}, HTTP_AUTHORIZATION=self.auth)
self.expect_error(status.HTTP_400_BAD_REQUEST, 'course_id_not_valid_ccx_id', resp)
# use an non existing ccx id
url = reverse('ccx_api:v0:ccx:detail', kwargs={'ccx_course_id': 'ccx-v1:foo.0+course_bar_0+Run_0+ccx@1'})
resp = client_request(url, {}, HTTP_AUTHORIZATION=self.auth)
self.expect_error(status.HTTP_404_NOT_FOUND, 'ccx_course_id_does_not_exist', resp)
# get a valid ccx key and add few 0s to get a non existing ccx for a valid course
ccx_key_str = '{0}000000'.format(self.ccx_key_str)
url = reverse('ccx_api:v0:ccx:detail', kwargs={'ccx_course_id': ccx_key_str})
resp = client_request(url, {}, HTTP_AUTHORIZATION=self.auth)
self.expect_error(status.HTTP_404_NOT_FOUND, 'ccx_course_id_does_not_exist', resp)
def test_get_detail(self):
"""
Test for getting detail of a ccx course
"""
resp = self.client.get(self.detail_url, {}, HTTP_AUTHORIZATION=self.auth)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
self.assertEqual(resp.data.get('ccx_course_id'), self.ccx_key_str) # pylint: disable=no-member
self.assertEqual(resp.data.get('display_name'), self.ccx.display_name) # pylint: disable=no-member
self.assertEqual(
resp.data.get('max_students_allowed'), # pylint: disable=no-member
self.ccx.max_student_enrollments_allowed # pylint: disable=no-member
)
self.assertEqual(resp.data.get('coach_email'), self.ccx.coach.email) # pylint: disable=no-member
self.assertEqual(resp.data.get('master_course_id'), unicode(self.ccx.course_id)) # pylint: disable=no-member
def test_delete_detail(self):
"""
Test for deleting a ccx course
"""
# check that there are overrides
self.assertGreater(CcxFieldOverride.objects.filter(ccx=self.ccx).count(), 0)
self.assertGreater(CourseEnrollment.objects.filter(course_id=self.ccx_key).count(), 0)
resp = self.client.delete(self.detail_url, {}, HTTP_AUTHORIZATION=self.auth)
self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
self.assertIsNone(resp.data) # pylint: disable=no-member
# the CCX does not exist any more
with self.assertRaises(CustomCourseForEdX.DoesNotExist):
CustomCourseForEdX.objects.get(id=self.ccx.id)
# check that there are no overrides
self.assertEqual(CcxFieldOverride.objects.filter(ccx=self.ccx).count(), 0)
self.assertEqual(CourseEnrollment.objects.filter(course_id=self.ccx_key).count(), 0)
def test_patch_detail_change_master_course(self):
"""
Test to patch a ccx course to change a master course
"""
data = {
'master_course_id': 'changed_course_id'
}
resp = self.client.patch(self.detail_url, data, format='json', HTTP_AUTHORIZATION=self.auth)
self.expect_error(status.HTTP_403_FORBIDDEN, 'master_course_id_change_not_allowed', resp)
@ddt.data(
(
{
'max_students_allowed': None,
'display_name': None,
'coach_email': None
},
{
'max_students_allowed': 'null_field_max_students_allowed',
'display_name': 'null_field_display_name',
'coach_email': 'null_field_coach_email'
}
),
(
{
'max_students_allowed': 10,
'display_name': 'CCX Title',
'coach_email': 'this is not an email@test.com'
},
{'coach_email': 'invalid_coach_email'}
),
(
{
'max_students_allowed': 10,
'display_name': '',
'coach_email': 'email@test.com'
},
{'display_name': 'invalid_display_name'}
),
(
{
'max_students_allowed': 'a',
'display_name': 'CCX Title',
'coach_email': 'email@test.com'
},
{'max_students_allowed': 'invalid_max_students_allowed'}
),
)
@ddt.unpack
def test_patch_detail_wrong_input_data(self, data, expected_errors):
"""
Test for different wrong inputs for the patch method
"""
resp = self.client.patch(self.detail_url, data, format='json', HTTP_AUTHORIZATION=self.auth)
self.expect_error_fields(expected_errors, resp)
def test_empty_patch(self):
"""
An empty patch does not modify anything
"""
display_name = self.ccx.display_name
max_students_allowed = self.ccx.max_student_enrollments_allowed # pylint: disable=no-member
coach_email = self.ccx.coach.email # pylint: disable=no-member
resp = self.client.patch(self.detail_url, {}, format='json', HTTP_AUTHORIZATION=self.auth)
self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
ccx = CustomCourseForEdX.objects.get(id=self.ccx.id)
self.assertEqual(display_name, ccx.display_name)
self.assertEqual(max_students_allowed, ccx.max_student_enrollments_allowed)
self.assertEqual(coach_email, ccx.coach.email)
def test_patch_detail_coach_does_not_exist(self):
"""
Specific test for the case when the input data is valid but the coach does not exist.
"""
data = {
'max_students_allowed': 111,
'display_name': 'CCX Title',
'coach_email': 'inexisting_email@test.com'
}
resp = self.client.patch(self.detail_url, data, format='json', HTTP_AUTHORIZATION=self.auth)
self.expect_error(status.HTTP_404_NOT_FOUND, 'coach_user_does_not_exist', resp)
def test_patch_detail(self):
"""
Test for successful patch
"""
outbox = self.get_outbox()
# create a new coach
new_coach = AdminFactory.create()
data = {
'max_students_allowed': 111,
'display_name': 'CCX Title',
'coach_email': new_coach.email
}
resp = self.client.patch(self.detail_url, data, format='json', HTTP_AUTHORIZATION=self.auth)
self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
ccx_from_db = CustomCourseForEdX.objects.get(id=self.ccx.id)
self.assertEqual(ccx_from_db.max_student_enrollments_allowed, data['max_students_allowed'])
self.assertEqual(ccx_from_db.display_name, data['display_name'])
self.assertEqual(ccx_from_db.coach.email, data['coach_email'])
# check that the coach user has coach role on the master course
coach_role_on_master_course = CourseCcxCoachRole(self.master_course_key)
self.assertTrue(coach_role_on_master_course.has_user(new_coach))
# check that the coach has been enrolled in the ccx
ccx_course_object = courses.get_course_by_id(self.ccx_key)
self.assertTrue(
CourseEnrollment.objects.filter(course_id=ccx_course_object.id, user=new_coach).exists()
)
# check that an email has been sent to the coach
self.assertEqual(len(outbox), 1)
self.assertIn(new_coach.email, outbox[0].recipients()) # pylint: disable=no-member
""" CCX API v0 URLs. """
from django.conf import settings
from django.conf.urls import patterns, url, include
from lms.djangoapps.ccx.api.v0 import views
CCX_COURSE_ID_PATTERN = settings.COURSE_ID_PATTERN.replace('course_id', 'ccx_course_id')
CCX_URLS = patterns(
'',
url(r'^$', views.CCXListView.as_view(), name='list'),
url(r'^{}/?$'.format(CCX_COURSE_ID_PATTERN), views.CCXDetailView.as_view(), name='detail'),
)
urlpatterns = patterns(
'',
url(r'^ccx/', include(CCX_URLS, namespace='ccx')),
)
""" API v0 views. """
import datetime
import logging
import pytz
from django.contrib.auth.models import User
from django.db import transaction
from django.http import Http404
from rest_framework import status
from rest_framework.authentication import SessionAuthentication
from rest_framework.generics import GenericAPIView
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework_oauth.authentication import OAuth2Authentication
from ccx_keys.locator import CCXLocator
from courseware import courses
from instructor.enrollment import (
enroll_email,
get_email_params,
)
from opaque_keys import InvalidKeyError
from opaque_keys.edx.keys import CourseKey
from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
from openedx.core.lib.api.permissions import IsCourseInstructor
from student.models import CourseEnrollment
from student.roles import CourseCcxCoachRole
from lms.djangoapps.ccx.models import CcxFieldOverride, CustomCourseForEdX
from lms.djangoapps.ccx.overrides import (
override_field_for_ccx,
)
from lms.djangoapps.ccx.utils import (
assign_coach_role_to_ccx,
is_email,
)
from .paginators import CCXAPIPagination
from .serializers import CCXCourseSerializer
log = logging.getLogger(__name__)
TODAY = datetime.datetime.today # for patching in tests
def get_valid_course(course_id, is_ccx=False, advanced_course_check=False):
"""
Helper function used to validate and get a course from a course_id string.
It works with both master and ccx course id.
Args:
course_id (str): A string representation of a Master or CCX Course ID.
is_ccx (bool): Flag to perform the right validation
advanced_course_check (bool): Flag to perform extra validations for the master course
Returns:
tuple: a tuple of course_object, course_key, error_code, http_status_code
"""
if course_id is None:
# the ccx detail view cannot call this function with a "None" value
# so the following `error_code` should be never used, but putting it
# to avoid a `NameError` exception in case this function will be used
# elsewhere in the future
error_code = 'course_id_not_provided'
if not is_ccx:
log.info('Master course ID not provided')
error_code = 'master_course_id_not_provided'
return None, None, error_code, status.HTTP_400_BAD_REQUEST
try:
course_key = CourseKey.from_string(course_id)
except InvalidKeyError:
log.info('Course ID string "%s" is not valid', course_id)
return None, None, 'course_id_not_valid', status.HTTP_400_BAD_REQUEST
if not is_ccx:
try:
course_object = courses.get_course_by_id(course_key)
except Http404:
log.info('Master Course with ID "%s" not found', course_id)
return None, None, 'course_id_does_not_exist', status.HTTP_404_NOT_FOUND
if advanced_course_check:
if course_object.id.deprecated:
return None, None, 'deprecated_master_course_id', status.HTTP_400_BAD_REQUEST
if not course_object.enable_ccx:
return None, None, 'ccx_not_enabled_for_master_course', status.HTTP_403_FORBIDDEN
return course_object, course_key, None, None
else:
try:
ccx_id = course_key.ccx
except AttributeError:
log.info('Course ID string "%s" is not a valid CCX ID', course_id)
return None, None, 'course_id_not_valid_ccx_id', status.HTTP_400_BAD_REQUEST
# get the master_course key
master_course_key = course_key.to_course_locator()
try:
ccx_course = CustomCourseForEdX.objects.get(id=ccx_id, course_id=master_course_key)
return ccx_course, course_key, None, None
except CustomCourseForEdX.DoesNotExist:
log.info('CCX Course with ID "%s" not found', course_id)
return None, None, 'ccx_course_id_does_not_exist', status.HTTP_404_NOT_FOUND
def get_valid_input(request_data, ignore_missing=False):
"""
Helper function to validate the data sent as input and to
build field based errors.
Args:
request_data (OrderedDict): the request data object
ignore_missing (bool): whether or not to ignore fields
missing from the input data
Returns:
tuple: a tuple of two dictionaries for valid input and field errors
"""
valid_input = {}
field_errors = {}
mandatory_fields = ('coach_email', 'display_name', 'max_students_allowed',)
# checking first if all the fields are present and they are not null
if not ignore_missing:
for field in mandatory_fields:
if field not in request_data:
field_errors[field] = {'error_code': 'missing_field_{0}'.format(field)}
if field_errors:
return valid_input, field_errors
# at this point I can assume that if the fields are present,
# they must be validated, otherwise they can be skipped
coach_email = request_data.get('coach_email')
if coach_email is not None:
if is_email(coach_email):
valid_input['coach_email'] = coach_email
else:
field_errors['coach_email'] = {'error_code': 'invalid_coach_email'}
elif 'coach_email' in request_data:
field_errors['coach_email'] = {'error_code': 'null_field_coach_email'}
display_name = request_data.get('display_name')
if display_name is not None:
if not display_name:
field_errors['display_name'] = {'error_code': 'invalid_display_name'}
else:
valid_input['display_name'] = display_name
elif 'display_name' in request_data:
field_errors['display_name'] = {'error_code': 'null_field_display_name'}
max_students_allowed = request_data.get('max_students_allowed')
if max_students_allowed is not None:
try:
max_students_allowed = int(max_students_allowed)
valid_input['max_students_allowed'] = max_students_allowed
except (TypeError, ValueError):
field_errors['max_students_allowed'] = {'error_code': 'invalid_max_students_allowed'}
elif 'max_students_allowed' in request_data:
field_errors['max_students_allowed'] = {'error_code': 'null_field_max_students_allowed'}
return valid_input, field_errors
def make_user_coach(user, master_course_key):
"""
Makes an user coach on the master course.
This function is needed because an user cannot become a coach of the CCX if s/he is not
coach on the master course.
Args:
user (User): User object
master_course_key (CourseKey): Key locator object for the course
"""
coach_role_on_master_course = CourseCcxCoachRole(master_course_key)
coach_role_on_master_course.add_users(user)
class CCXListView(GenericAPIView):
"""
**Use Case**
* Get the list of CCX courses for a given master course.
* Creates a new CCX course for a given master course.
**Example Request**
GET /api/ccx/v0/ccx/?master_course_id={master_course_id}
POST /api/ccx/v0/ccx {
"master_course_id": "course-v1:Organization+EX101+RUN-FALL2099",
"display_name": "CCX example title",
"coach_email": "john@example.com",
"max_students_allowed": 123
}
**GET Parameters**
A GET request can include the following parameters.
* master_course_id: A string representation of a Master Course ID. Note that this must be properly
encoded by the client.
* page: Optional. An integer representing the pagination instance number.
* order_by: Optional. A string representing the field by which sort the results.
* sort_order: Optional. A string (either "asc" or "desc") indicating the desired order.
**POST Parameters**
A POST request can include the following parameters.
* master_course_id: A string representation of a Master Course ID.
* display_name: A string representing the CCX Course title.
* coach_email: A string representing the CCX owner email.
* max_students_allowed: An integer representing he maximum number of students that
can be enrolled in the CCX Course.
**GET Response Values**
If the request for information about the course is successful, an HTTP 200 "OK" response
is returned with a collection of CCX courses for the specified master course.
The HTTP 200 response has the following values.
* results: a collection of CCX courses. Each CCX course contains the following values:
* ccx_course_id: A string representation of a CCX Course ID.
* display_name: A string representing the CCX Course title.
* coach_email: A string representing the CCX owner email.
* start: A string representing the start date for the CCX Course.
* due: A string representing the due date for the CCX Course.
* max_students_allowed: An integer representing he maximum number of students that
can be enrolled in the CCX Course.
* count: An integer representing the total number of records that matched the request parameters.
* next: A string representing the URL where to retrieve the next page of results. This can be `null`
in case the response contains the complete list of results.
* previous: A string representing the URL where to retrieve the previous page of results. This can be
`null` in case the response contains the first page of results.
**Example GET Response**
{
"count": 99,
"next": "https://openedx-ccx-api-instance.org/api/ccx/v0/ccx/?course_id=<course_id>&page=2",
"previous": null,
"results": {
{
"ccx_course_id": "ccx-v1:Organization+EX101+RUN-FALL2099+ccx@1",
"display_name": "CCX example title",
"coach_email": "john@example.com",
"start": "2019-01-01",
"due": "2019-06-01",
"max_students_allowed": 123
},
{ ... }
}
}
**POST Response Values**
If the request for the creation of a CCX Course is successful, an HTTP 201 "Created" response
is returned with the newly created CCX details.
The HTTP 201 response has the following values.
* ccx_course_id: A string representation of a CCX Course ID.
* display_name: A string representing the CCX Course title.
* coach_email: A string representing the CCX owner email.
* start: A string representing the start date for the CCX Course.
* due: A string representing the due date for the CCX Course.
* max_students_allowed: An integer representing he maximum number of students that
can be enrolled in the CCX Course.
**Example POST Response**
{
"ccx_course_id": "ccx-v1:Organization+EX101+RUN-FALL2099+ccx@1",
"display_name": "CCX example title",
"coach_email": "john@example.com",
"start": "2019-01-01",
"due": "2019-06-01",
"max_students_allowed": 123
}
"""
authentication_classes = (OAuth2Authentication, SessionAuthentication,)
permission_classes = (IsAuthenticated, IsCourseInstructor)
serializer_class = CCXCourseSerializer
pagination_class = CCXAPIPagination
def get(self, request):
"""
Gets a list of CCX Courses for a given Master Course.
Additional parameters are allowed for pagination purposes.
Args:
request (Request): Django request object.
Return:
A JSON serialized representation of a list of CCX courses.
"""
master_course_id = request.GET.get('master_course_id')
master_course_object, master_course_key, error_code, http_status = get_valid_course(master_course_id)
if master_course_object is None:
return Response(
status=http_status,
data={
'error_code': error_code
}
)
queryset = CustomCourseForEdX.objects.filter(course_id=master_course_key)
order_by_input = request.query_params.get('order_by')
sort_order_input = request.query_params.get('sort_order')
if order_by_input in ('id', 'display_name'):
sort_direction = ''
if sort_order_input == 'desc':
sort_direction = '-'
queryset = queryset.order_by('{0}{1}'.format(sort_direction, order_by_input))
page = self.paginate_queryset(queryset)
serializer = self.get_serializer(page, many=True)
response = self.get_paginated_response(serializer.data)
return response
def post(self, request):
"""
Creates a new CCX course for a given Master Course.
Args:
request (Request): Django request object.
Return:
A JSON serialized representation a newly created CCX course.
"""
master_course_id = request.data.get('master_course_id')
master_course_object, master_course_key, error_code, http_status = get_valid_course(
master_course_id,
advanced_course_check=True
)
if master_course_object is None:
return Response(
status=http_status,
data={
'error_code': error_code
}
)
# validating the rest of the input
valid_input, field_errors = get_valid_input(request.data)
if field_errors:
return Response(
status=status.HTTP_400_BAD_REQUEST,
data={
'field_errors': field_errors
}
)
try:
coach = User.objects.get(email=valid_input['coach_email'])
except User.DoesNotExist:
return Response(
status=status.HTTP_404_NOT_FOUND,
data={
'error_code': 'coach_user_does_not_exist'
}
)
with transaction.atomic():
ccx_course_object = CustomCourseForEdX(
course_id=master_course_object.id,
coach=coach,
display_name=valid_input['display_name'])
ccx_course_object.save()
# Make sure start/due are overridden for entire course
start = TODAY().replace(tzinfo=pytz.UTC)
override_field_for_ccx(ccx_course_object, master_course_object, 'start', start)
override_field_for_ccx(ccx_course_object, master_course_object, 'due', None)
# Enforce a static limit for the maximum amount of students that can be enrolled
override_field_for_ccx(
ccx_course_object,
master_course_object,
'max_student_enrollments_allowed',
valid_input['max_students_allowed']
)
# Hide anything that can show up in the schedule
hidden = 'visible_to_staff_only'
for chapter in master_course_object.get_children():
override_field_for_ccx(ccx_course_object, chapter, hidden, True)
for sequential in chapter.get_children():
override_field_for_ccx(ccx_course_object, sequential, hidden, True)
for vertical in sequential.get_children():
override_field_for_ccx(ccx_course_object, vertical, hidden, True)
# make the coach user a coach on the master course
make_user_coach(coach, master_course_key)
# pull the ccx course key
ccx_course_key = CCXLocator.from_course_locator(master_course_object.id, ccx_course_object.id)
# enroll the coach in the newly created ccx
email_params = get_email_params(
master_course_object,
auto_enroll=True,
course_key=ccx_course_key,
display_name=ccx_course_object.display_name
)
enroll_email(
course_id=ccx_course_key,
student_email=coach.email,
auto_enroll=True,
email_students=True,
email_params=email_params,
)
# assign coach role for the coach to the newly created ccx
assign_coach_role_to_ccx(ccx_course_key, coach, master_course_object.id)
serializer = self.get_serializer(ccx_course_object)
return Response(
status=status.HTTP_201_CREATED,
data=serializer.data
)
class CCXDetailView(GenericAPIView):
"""
**Use Case**
* Get the details of CCX course.
* Modify a CCX course.
* Delete a CCX course.
**Example Request**
GET /api/ccx/v0/ccx/{ccx_course_id}
PATCH /api/ccx/v0/ccx/{ccx_course_id} {
"display_name": "CCX example title modified",
"coach_email": "joe@example.com",
"max_students_allowed": 111
}
DELETE /api/ccx/v0/ccx/{ccx_course_id}
**GET and DELETE Parameters**
A GET or DELETE request must include the following parameter.
* ccx_course_id: A string representation of a CCX Course ID.
**PATCH Parameters**
A PATCH request can include the following parameters
* ccx_course_id: A string representation of a CCX Course ID.
* display_name: Optional. A string representing the CCX Course title.
* coach_email: Optional. A string representing the CCX owner email.
* max_students_allowed: Optional. An integer representing he maximum number of students that
can be enrolled in the CCX Course.
**GET Response Values**
If the request for information about the CCX course is successful, an HTTP 200 "OK" response
is returned.
The HTTP 200 response has the following values.
* ccx_course_id: A string representation of a CCX Course ID.
* display_name: A string representing the CCX Course title.
* coach_email: A string representing the CCX owner email.
* start: A string representing the start date for the CCX Course.
* due: A string representing the due date for the CCX Course.
* max_students_allowed: An integer representing he maximum number of students that
can be enrolled in the CCX Course.
**PATCH and DELETE Response Values**
If the request for modification or deletion of a CCX course is successful, an HTTP 204 "No Content"
response is returned.
"""
authentication_classes = (OAuth2Authentication, SessionAuthentication,)
permission_classes = (IsAuthenticated, IsCourseInstructor)
serializer_class = CCXCourseSerializer
def get(self, request, ccx_course_id=None):
"""
Gets a CCX Course information.
Args:
request (Request): Django request object.
ccx_course_id (string): URI element specifying the CCX course location.
Return:
A JSON serialized representation of the CCX course.
"""
ccx_course_object, _, error_code, http_status = get_valid_course(ccx_course_id, is_ccx=True)
if ccx_course_object is None:
return Response(
status=http_status,
data={
'error_code': error_code
}
)
serializer = self.get_serializer(ccx_course_object)
return Response(serializer.data)
def delete(self, request, ccx_course_id=None): # pylint: disable=unused-argument
"""
Deletes a CCX course.
Args:
request (Request): Django request object.
ccx_course_id (string): URI element specifying the CCX course location.
"""
ccx_course_object, ccx_course_key, error_code, http_status = get_valid_course(ccx_course_id, is_ccx=True)
if ccx_course_object is None:
return Response(
status=http_status,
data={
'error_code': error_code
}
)
ccx_course_overview = CourseOverview.get_from_id(ccx_course_key)
# clean everything up with a single transaction
with transaction.atomic():
CcxFieldOverride.objects.filter(ccx=ccx_course_object).delete()
# remove all users enrolled in the CCX from the CourseEnrollment model
CourseEnrollment.objects.filter(course_id=ccx_course_key).delete()
ccx_course_overview.delete()
ccx_course_object.delete()
return Response(
status=status.HTTP_204_NO_CONTENT,
)
def patch(self, request, ccx_course_id=None):
"""
Modifies a CCX course.
Args:
request (Request): Django request object.
ccx_course_id (string): URI element specifying the CCX course location.
"""
ccx_course_object, ccx_course_key, error_code, http_status = get_valid_course(ccx_course_id, is_ccx=True)
if ccx_course_object is None:
return Response(
status=http_status,
data={
'error_code': error_code
}
)
master_course_id = request.data.get('master_course_id')
if master_course_id is not None and unicode(ccx_course_object.course_id) != master_course_id:
return Response(
status=status.HTTP_403_FORBIDDEN,
data={
'error_code': 'master_course_id_change_not_allowed'
}
)
valid_input, field_errors = get_valid_input(request.data, ignore_missing=True)
if field_errors:
return Response(
status=status.HTTP_400_BAD_REQUEST,
data={
'field_errors': field_errors
}
)
with transaction.atomic():
# update the display name
if 'display_name' in valid_input:
ccx_course_object.display_name = valid_input['display_name']
# check if the coach has changed and in case update it
old_coach = None
if 'coach_email' in valid_input:
try:
coach = User.objects.get(email=valid_input['coach_email'])
except User.DoesNotExist:
return Response(
status=status.HTTP_404_NOT_FOUND,
data={
'error_code': 'coach_user_does_not_exist'
}
)
if ccx_course_object.coach.id != coach.id:
old_coach = ccx_course_object.coach
ccx_course_object.coach = coach
ccx_course_object.save()
# update the overridden field for the maximum amount of students
if 'max_students_allowed' in valid_input:
override_field_for_ccx(
ccx_course_object,
ccx_course_object.course,
'max_student_enrollments_allowed',
valid_input['max_students_allowed']
)
# if the coach has changed, update the permissions
if old_coach is not None:
# get the master course key and master course object
master_course_object, master_course_key, _, _ = get_valid_course(unicode(ccx_course_object.course_id))
# make the new ccx coach a coach on the master course
make_user_coach(coach, master_course_key)
# enroll the coach in the ccx
email_params = get_email_params(
master_course_object,
auto_enroll=True,
course_key=ccx_course_key,
display_name=ccx_course_object.display_name
)
enroll_email(
course_id=ccx_course_key,
student_email=coach.email,
auto_enroll=True,
email_students=True,
email_params=email_params,
)
# enroll the coach to the newly created ccx
assign_coach_role_to_ccx(ccx_course_key, coach, master_course_object.id)
return Response(
status=status.HTTP_204_NO_CONTENT,
)
""" """
Dummy factories for tests Dummy factories for tests
""" """
from factory import SubFactory from factory import SubFactory, Sequence
from factory.django import DjangoModelFactory from factory.django import DjangoModelFactory
from student.tests.factories import UserFactory from student.tests.factories import UserFactory
from lms.djangoapps.ccx.models import CustomCourseForEdX from lms.djangoapps.ccx.models import CustomCourseForEdX
...@@ -11,6 +11,6 @@ class CcxFactory(DjangoModelFactory): # pylint: disable=missing-docstring ...@@ -11,6 +11,6 @@ class CcxFactory(DjangoModelFactory): # pylint: disable=missing-docstring
class Meta(object): class Meta(object):
model = CustomCourseForEdX model = CustomCourseForEdX
display_name = "Test CCX" display_name = Sequence(lambda n: 'Test CCX #{0}'.format(n)) # pylint: disable=unnecessary-lambda
id = None # pylint: disable=invalid-name id = None # pylint: disable=invalid-name
coach = SubFactory(UserFactory) coach = SubFactory(UserFactory)
...@@ -20,7 +20,7 @@ from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory ...@@ -20,7 +20,7 @@ from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory
from lms.djangoapps.ccx.models import CustomCourseForEdX from lms.djangoapps.ccx.models import CustomCourseForEdX
from lms.djangoapps.ccx.overrides import override_field_for_ccx from lms.djangoapps.ccx.overrides import override_field_for_ccx
from lms.djangoapps.ccx.tests.test_views import flatten, iter_blocks from lms.djangoapps.ccx.tests.utils import flatten, iter_blocks
@attr('shard_1') @attr('shard_1')
......
...@@ -16,8 +16,6 @@ from courseware.tests.factories import StudentModuleFactory ...@@ -16,8 +16,6 @@ from courseware.tests.factories import StudentModuleFactory
from courseware.tests.helpers import LoginEnrollmentTestCase from courseware.tests.helpers import LoginEnrollmentTestCase
from courseware.tabs import get_course_tab_list from courseware.tabs import get_course_tab_list
from django.conf import settings from django.conf import settings
from django.core.exceptions import ValidationError
from django.core.validators import validate_email
from django.core.urlresolvers import reverse, resolve from django.core.urlresolvers import reverse, resolve
from django.utils.timezone import UTC from django.utils.timezone import UTC
from django.test.utils import override_settings from django.test.utils import override_settings
...@@ -52,6 +50,11 @@ from ccx_keys.locator import CCXLocator ...@@ -52,6 +50,11 @@ from ccx_keys.locator import CCXLocator
from lms.djangoapps.ccx.models import CustomCourseForEdX from lms.djangoapps.ccx.models import CustomCourseForEdX
from lms.djangoapps.ccx.overrides import get_override_for_ccx, override_field_for_ccx from lms.djangoapps.ccx.overrides import get_override_for_ccx, override_field_for_ccx
from lms.djangoapps.ccx.tests.factories import CcxFactory from lms.djangoapps.ccx.tests.factories import CcxFactory
from lms.djangoapps.ccx.tests.utils import (
CcxTestCase,
flatten,
)
from lms.djangoapps.ccx.utils import is_email
from lms.djangoapps.ccx.views import get_date from lms.djangoapps.ccx.views import get_date
...@@ -114,96 +117,24 @@ def setup_students_and_grades(context): ...@@ -114,96 +117,24 @@ def setup_students_and_grades(context):
) )
def is_email(identifier):
"""
Checks if an `identifier` string is a valid email
"""
try:
validate_email(identifier)
except ValidationError:
return False
return True
@attr('shard_1') @attr('shard_1')
@ddt.ddt @ddt.ddt
class TestCoachDashboard(SharedModuleStoreTestCase, LoginEnrollmentTestCase): class TestCoachDashboard(CcxTestCase, LoginEnrollmentTestCase):
""" """
Tests for Custom Courses views. Tests for Custom Courses views.
""" """
MODULESTORE = TEST_DATA_SPLIT_MODULESTORE
@classmethod @classmethod
def setUpClass(cls): def setUpClass(cls):
super(TestCoachDashboard, cls).setUpClass() super(TestCoachDashboard, cls).setUpClass()
cls.course = course = CourseFactory.create()
# Create a course outline
cls.mooc_start = start = datetime.datetime(
2010, 5, 12, 2, 42, tzinfo=pytz.UTC
)
cls.mooc_due = due = datetime.datetime(
2010, 7, 7, 0, 0, tzinfo=pytz.UTC
)
cls.chapters = [
ItemFactory.create(start=start, parent=course) for _ in xrange(2)
]
cls.sequentials = flatten([
[
ItemFactory.create(parent=chapter) for _ in xrange(2)
] for chapter in cls.chapters
])
cls.verticals = flatten([
[
ItemFactory.create(
start=start, due=due, parent=sequential, graded=True, format='Homework', category=u'vertical'
) for _ in xrange(2)
] for sequential in cls.sequentials
])
# Trying to wrap the whole thing in a bulk operation fails because it
# doesn't find the parents. But we can at least wrap this part...
with cls.store.bulk_operations(course.id, emit_signals=False):
blocks = flatten([ # pylint: disable=unused-variable
[
ItemFactory.create(parent=vertical) for _ in xrange(2)
] for vertical in cls.verticals
])
def setUp(self): def setUp(self):
""" """
Set up tests Set up tests
""" """
super(TestCoachDashboard, self).setUp() super(TestCoachDashboard, self).setUp()
# Login with the instructor account
# Create instructor account self.client.login(username=self.coach.username, password="test")
self.coach = coach = AdminFactory.create()
self.client.login(username=coach.username, password="test")
# create an instance of modulestore
self.mstore = modulestore()
def make_coach(self):
"""
create coach user
"""
role = CourseCcxCoachRole(self.course.id)
role.add_users(self.coach)
def make_ccx(self, max_students_allowed=settings.CCX_MAX_STUDENTS_ALLOWED):
"""
create ccx
"""
ccx = CcxFactory(course_id=self.course.id, coach=self.coach)
override_field_for_ccx(ccx, self.course, 'max_student_enrollments_allowed', max_students_allowed)
return ccx
def get_outbox(self):
"""
get fake outbox
"""
from django.core import mail
return mail.outbox
def assert_elements_in_schedule(self, url, n_chapters=2, n_sequentials=4, n_verticals=8): def assert_elements_in_schedule(self, url, n_chapters=2, n_sequentials=4, n_verticals=8):
""" """
...@@ -1005,23 +936,3 @@ class TestStudentDashboardWithCCX(ModuleStoreTestCase): ...@@ -1005,23 +936,3 @@ class TestStudentDashboardWithCCX(ModuleStoreTestCase):
response = self.client.get(reverse('dashboard')) response = self.client.get(reverse('dashboard'))
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
self.assertTrue(re.search('Test CCX', response.content)) self.assertTrue(re.search('Test CCX', response.content))
def flatten(seq):
"""
For [[1, 2], [3, 4]] returns [1, 2, 3, 4]. Does not recurse.
"""
return [x for sub in seq for x in sub]
def iter_blocks(course):
"""
Returns an iterator over all of the blocks in a course.
"""
def visit(block):
""" get child blocks """
yield block
for child in block.get_children():
for descendant in visit(child): # wish they'd backport yield from
yield descendant
return visit(course)
"""
Test utils for CCX
"""
import datetime
import pytz
from django.conf import settings
from lms.djangoapps.ccx.overrides import override_field_for_ccx
from lms.djangoapps.ccx.tests.factories import CcxFactory
from student.roles import CourseCcxCoachRole
from student.tests.factories import (
AdminFactory,
)
from xmodule.modulestore.django import modulestore
from xmodule.modulestore.tests.django_utils import (
SharedModuleStoreTestCase,
TEST_DATA_SPLIT_MODULESTORE
)
from xmodule.modulestore.tests.factories import (
CourseFactory,
ItemFactory,
)
class CcxTestCase(SharedModuleStoreTestCase):
"""
General test class to be used in other CCX tests classes.
It creates a course that can be used as master course for CCXs.
"""
MODULESTORE = TEST_DATA_SPLIT_MODULESTORE
@classmethod
def setUpClass(cls):
super(CcxTestCase, cls).setUpClass()
cls.course = course = CourseFactory.create()
# Create a course outline
cls.mooc_start = start = datetime.datetime(
2010, 5, 12, 2, 42, tzinfo=pytz.UTC
)
cls.mooc_due = due = datetime.datetime(
2010, 7, 7, 0, 0, tzinfo=pytz.UTC
)
cls.chapters = [
ItemFactory.create(start=start, parent=course) for _ in xrange(2)
]
cls.sequentials = flatten([
[
ItemFactory.create(parent=chapter) for _ in xrange(2)
] for chapter in cls.chapters
])
cls.verticals = flatten([
[
ItemFactory.create(
start=start, due=due, parent=sequential, graded=True, format='Homework', category=u'vertical'
) for _ in xrange(2)
] for sequential in cls.sequentials
])
# Trying to wrap the whole thing in a bulk operation fails because it
# doesn't find the parents. But we can at least wrap this part...
with cls.store.bulk_operations(course.id, emit_signals=False):
blocks = flatten([ # pylint: disable=unused-variable
[
ItemFactory.create(parent=vertical) for _ in xrange(2)
] for vertical in cls.verticals
])
def setUp(self):
"""
Set up tests
"""
super(CcxTestCase, self).setUp()
# Create instructor account
self.coach = AdminFactory.create()
# create an instance of modulestore
self.mstore = modulestore()
def make_coach(self):
"""
create coach user
"""
role = CourseCcxCoachRole(self.course.id)
role.add_users(self.coach)
def make_ccx(self, max_students_allowed=settings.CCX_MAX_STUDENTS_ALLOWED):
"""
create ccx
"""
ccx = CcxFactory(course_id=self.course.id, coach=self.coach)
override_field_for_ccx(ccx, self.course, 'max_student_enrollments_allowed', max_students_allowed)
return ccx
def get_outbox(self):
"""
get fake outbox
"""
from django.core import mail
return mail.outbox
def flatten(seq):
"""
For [[1, 2], [3, 4]] returns [1, 2, 3, 4]. Does not recurse.
"""
return [x for sub in seq for x in sub]
def iter_blocks(course):
"""
Returns an iterator over all of the blocks in a course.
"""
def visit(block):
""" get child blocks """
yield block
for child in block.get_children():
for descendant in visit(child): # wish they'd backport yield from
yield descendant
return visit(course)
...@@ -6,14 +6,12 @@ Does not include any access control, be sure to check access before calling. ...@@ -6,14 +6,12 @@ Does not include any access control, be sure to check access before calling.
import datetime import datetime
import logging import logging
import pytz import pytz
from contextlib import contextmanager from contextlib import contextmanager
from django.contrib.auth.models import User from django.contrib.auth.models import User
from django.core.exceptions import ValidationError from django.core.exceptions import ValidationError
from django.utils.translation import ugettext as _ from django.utils.translation import ugettext as _
from django.core.validators import validate_email from django.core.validators import validate_email
from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
from courseware.courses import get_course_by_id from courseware.courses import get_course_by_id
from courseware.model_data import FieldDataCache from courseware.model_data import FieldDataCache
...@@ -24,6 +22,7 @@ from instructor.enrollment import ( ...@@ -24,6 +22,7 @@ from instructor.enrollment import (
) )
from instructor.access import allow_access from instructor.access import allow_access
from instructor.views.tools import get_student_from_identifier from instructor.views.tools import get_student_from_identifier
from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
from student.models import CourseEnrollment from student.models import CourseEnrollment
from student.roles import CourseCcxCoachRole from student.roles import CourseCcxCoachRole
...@@ -250,3 +249,14 @@ def assign_coach_role_to_ccx(ccx_locator, user, master_course_id): ...@@ -250,3 +249,14 @@ def assign_coach_role_to_ccx(ccx_locator, user, master_course_id):
# assign user role coach on ccx # assign user role coach on ccx
with ccx_course(ccx_locator) as course: with ccx_course(ccx_locator) as course:
allow_access(course, user, "ccx_coach", send_email=False) allow_access(course, user, "ccx_coach", send_email=False)
def is_email(identifier):
"""
Checks if an `identifier` string is a valid email
"""
try:
validate_email(identifier)
except ValidationError:
return False
return True
...@@ -14,7 +14,7 @@ import pystache_custom as pystache ...@@ -14,7 +14,7 @@ import pystache_custom as pystache
from opaque_keys.edx.locations import i4xEncoder from opaque_keys.edx.locations import i4xEncoder
from opaque_keys.edx.keys import CourseKey from opaque_keys.edx.keys import CourseKey
from xmodule.modulestore.django import modulestore from xmodule.modulestore.django import modulestore
from ccx.overrides import get_current_ccx from lms.djangoapps.ccx.overrides import get_current_ccx
from django_comment_common.models import Role, FORUM_ROLE_STUDENT from django_comment_common.models import Role, FORUM_ROLE_STUDENT
from django_comment_client.permissions import check_permissions_by_view, has_permission, get_team from django_comment_client.permissions import check_permissions_by_view, has_permission, get_team
......
...@@ -10,7 +10,7 @@ from django.test.client import RequestFactory ...@@ -10,7 +10,7 @@ from django.test.client import RequestFactory
from django.test.utils import override_settings from django.test.utils import override_settings
from edxmako.shortcuts import render_to_response from edxmako.shortcuts import render_to_response
from ccx.tests.test_views import setup_students_and_grades from lms.djangoapps.ccx.tests.test_views import setup_students_and_grades
from courseware.tabs import get_course_tab_list from courseware.tabs import get_course_tab_list
from courseware.tests.factories import UserFactory from courseware.tests.factories import UserFactory
from courseware.tests.helpers import LoginEnrollmentTestCase from courseware.tests.helpers import LoginEnrollmentTestCase
......
...@@ -935,6 +935,7 @@ if settings.FEATURES["CUSTOM_COURSES_EDX"]: ...@@ -935,6 +935,7 @@ if settings.FEATURES["CUSTOM_COURSES_EDX"]:
urlpatterns += ( urlpatterns += (
url(r'^courses/{}/'.format(settings.COURSE_ID_PATTERN), url(r'^courses/{}/'.format(settings.COURSE_ID_PATTERN),
include('ccx.urls')), include('ccx.urls')),
url(r'^api/ccx/', include('lms.djangoapps.ccx.api.urls', namespace='ccx_api')),
) )
# Access to courseware as an LTI provider # Access to courseware as an LTI provider
......
...@@ -6,7 +6,7 @@ from django.conf import settings ...@@ -6,7 +6,7 @@ from django.conf import settings
from django.http import Http404 from django.http import Http404
from rest_framework import permissions from rest_framework import permissions
from student.roles import CourseStaffRole from student.roles import CourseStaffRole, CourseInstructorRole
class ApiKeyHeaderPermission(permissions.BasePermission): class ApiKeyHeaderPermission(permissions.BasePermission):
...@@ -64,6 +64,15 @@ class IsUserInUrl(permissions.BasePermission): ...@@ -64,6 +64,15 @@ class IsUserInUrl(permissions.BasePermission):
return True return True
class IsCourseInstructor(permissions.BasePermission):
"""
Permission to check that user is a course instructor.
"""
def has_object_permission(self, request, view, obj):
return hasattr(request, 'user') and CourseInstructorRole(obj.course_id).has_user(request.user)
class IsUserInUrlOrStaff(IsUserInUrl): class IsUserInUrlOrStaff(IsUserInUrl):
""" """
Permission that checks to see if the request user matches the user in the URL or has is_staff access. Permission that checks to see if the request user matches the user in the URL or has is_staff access.
......
...@@ -3,13 +3,48 @@ ...@@ -3,13 +3,48 @@
import ddt import ddt
from django.test import TestCase, RequestFactory from django.test import TestCase, RequestFactory
from openedx.core.lib.api.permissions import IsStaffOrOwner from student.roles import CourseStaffRole, CourseInstructorRole
from openedx.core.lib.api.permissions import IsStaffOrOwner, IsCourseInstructor
from student.tests.factories import UserFactory from student.tests.factories import UserFactory
from opaque_keys.edx.keys import CourseKey
class TestObject(object): class TestObject(object):
""" Fake class for object permission tests. """ """ Fake class for object permission tests. """
user = None def __init__(self, user=None, course_id=None):
self.user = user
self.course_id = course_id
class IsCourseInstructorTests(TestCase):
""" Test for IsCourseInstructor permission class. """
def setUp(self):
super(IsCourseInstructorTests, self).setUp()
self.permission = IsCourseInstructor()
self.request = RequestFactory().get('/')
self.course_key = CourseKey.from_string('edx/test123/run')
self.obj = TestObject(course_id=self.course_key)
def test_course_staff_has_no_access(self):
user = UserFactory.create()
self.request.user = user
CourseStaffRole(course_key=self.course_key).add_users(user)
self.assertFalse(
self.permission.has_object_permission(self.request, None, self.obj))
def test_course_instructor_has_access(self):
user = UserFactory.create()
self.request.user = user
CourseInstructorRole(course_key=self.course_key).add_users(user)
self.assertTrue(
self.permission.has_object_permission(self.request, None, self.obj))
def test_anonymous_has_no_access(self):
self.assertFalse(
self.permission.has_object_permission(self.request, None, self.obj))
@ddt.ddt @ddt.ddt
......
...@@ -10,7 +10,8 @@ import branding ...@@ -10,7 +10,8 @@ import branding
# app that handles site status messages # app that handles site status messages
from status.status import get_site_status_msg from status.status import get_site_status_msg
from ccx.overrides import get_current_ccx from microsite_configuration import microsite
from lms.djangoapps.ccx.overrides import get_current_ccx
%> %>
## Provide a hook for themes to inject branding on top. ## Provide a hook for themes to inject branding on top.
......
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