Merge pull request #1979 from edx/alex/lti_fixes_to_release
Alex/lti fixes to release
This commit is contained in:
@@ -5,13 +5,16 @@ These are notable changes in edx-platform. This is a rolling list of changes,
|
||||
in roughly chronological order, most recent first. Add your entries at or near
|
||||
the top. Include a label indicating the component affected.
|
||||
|
||||
Blades: Make LTI module not send grade_back_url if has_score=False. BLD-561.
|
||||
Blades: LTI additional Python tests. LTI must use HTTPS for
|
||||
lis_outcome_service_url. BLD-564.
|
||||
|
||||
Blades: Fix bug when Image mapping problems are not working for students in IE. BLD-413.
|
||||
|
||||
Blades: Add template that displays the most up-to-date features of
|
||||
drag-and-drop. BLD-479.
|
||||
|
||||
Blades: LTI additional Python tests. LTI fix bug e-reader error when popping
|
||||
out window. BLD-465.
|
||||
Blades: LTI fix bug e-reader error when popping out window. BLD-465.
|
||||
|
||||
Common: Switch from mitx.db to edx.db for sqlite databases. This will effectively
|
||||
reset state for local instances of the code, unless you manually rename your
|
||||
|
||||
@@ -46,8 +46,8 @@ class AnonymousUserId(models.Model):
|
||||
|
||||
Purpose of this table is to provide user by anonymous_user_id.
|
||||
|
||||
We are generating anonymous_user_id using md5 algorithm, so resulting length will always be 16 bytes.
|
||||
http://docs.python.org/2/library/md5.html#md5.digest_size
|
||||
We generate anonymous_user_id using md5 algorithm,
|
||||
and use result in hex form, so its length is equal to 32 bytes.
|
||||
"""
|
||||
user = models.ForeignKey(User, db_index=True)
|
||||
anonymous_user_id = models.CharField(unique=True, max_length=32)
|
||||
|
||||
@@ -259,37 +259,22 @@ class LTIModule(LTIFields, XModule):
|
||||
'element_class': self.category,
|
||||
'open_in_a_new_page': self.open_in_a_new_page,
|
||||
'display_name': self.display_name,
|
||||
'form_url': self.get_form_path(),
|
||||
'form_url': self.runtime.handler_url(self, 'preview_handler').rstrip('/?'),
|
||||
}
|
||||
|
||||
|
||||
def get_form_path(self):
|
||||
return self.runtime.handler_url(self, 'preview_handler').rstrip('/?')
|
||||
|
||||
def get_html(self):
|
||||
"""
|
||||
Renders parameters to template.
|
||||
"""
|
||||
return self.system.render_template('lti.html', self.get_context())
|
||||
|
||||
def get_form(self):
|
||||
"""
|
||||
Renders parameters to form template.
|
||||
"""
|
||||
return self.system.render_template('lti_form.html', self.get_context())
|
||||
|
||||
@XBlock.handler
|
||||
def preview_handler(self, request, dispatch):
|
||||
def preview_handler(self, _, __):
|
||||
"""
|
||||
Ajax handler.
|
||||
|
||||
Args:
|
||||
dispatch: string request slug
|
||||
|
||||
Returns:
|
||||
json string
|
||||
This is called to get context with new oauth params to iframe.
|
||||
"""
|
||||
return Response(self.get_form(), content_type='text/html')
|
||||
template = self.system.render_template('lti_form.html', self.get_context())
|
||||
return Response(template, content_type='text/html')
|
||||
|
||||
def get_user_id(self):
|
||||
user_id = self.runtime.anonymous_student_id
|
||||
@@ -299,11 +284,18 @@ class LTIModule(LTIFields, XModule):
|
||||
def get_outcome_service_url(self):
|
||||
"""
|
||||
Return URL for storing grades.
|
||||
|
||||
To test LTI on sandbox we must use http scheme.
|
||||
|
||||
While testing locally and on Jenkins, mock_lti_server use http.referer
|
||||
to obtain scheme, so it is ok to have http(s) anyway.
|
||||
"""
|
||||
uri = 'http://{host}{path}'.format(
|
||||
host=self.system.hostname,
|
||||
path=self.runtime.handler_url(self, 'grade_handler', thirdparty=True).rstrip('/?')
|
||||
)
|
||||
scheme = 'http' if 'sandbox' in self.system.hostname else 'https'
|
||||
uri = '{scheme}://{host}{path}'.format(
|
||||
scheme=scheme,
|
||||
host=self.system.hostname,
|
||||
path=self.runtime.handler_url(self, 'grade_handler', thirdparty=True).rstrip('/?')
|
||||
)
|
||||
return uri
|
||||
|
||||
def get_resource_link_id(self):
|
||||
@@ -363,11 +355,15 @@ class LTIModule(LTIFields, XModule):
|
||||
|
||||
# Parameters required for grading:
|
||||
u'resource_link_id': self.get_resource_link_id(),
|
||||
u'lis_outcome_service_url': self.get_outcome_service_url(),
|
||||
u'lis_result_sourcedid': self.get_lis_result_sourcedid(),
|
||||
|
||||
}
|
||||
|
||||
if self.has_score:
|
||||
body.update({
|
||||
u'lis_outcome_service_url': self.get_outcome_service_url()
|
||||
})
|
||||
|
||||
# Appending custom parameter for signing.
|
||||
body.update(custom_parameters)
|
||||
|
||||
@@ -449,7 +445,7 @@ oauth_consumer_key="", oauth_signature="frVp4JuvT1mVXlxktiAUjQ7%2F1cw%3D"'}
|
||||
|
||||
Example of correct/incorrect answer XML body:: see response_xml_template.
|
||||
"""
|
||||
response_xml_template = textwrap.dedent("""
|
||||
response_xml_template = textwrap.dedent("""\
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<imsx_POXEnvelopeResponse xmlns = "http://www.imsglobal.org/services/ltiv1p1/xsd/imsoms_v1p0">
|
||||
<imsx_POXHeader>
|
||||
@@ -491,6 +487,8 @@ oauth_consumer_key="", oauth_signature="frVp4JuvT1mVXlxktiAUjQ7%2F1cw%3D"'}
|
||||
try:
|
||||
imsx_messageIdentifier, sourcedId, score, action = self.parse_grade_xml_body(request.body)
|
||||
except Exception:
|
||||
log.debug("[LTI]: Request body XML parsing error.")
|
||||
failure_values['imsx_description'] = 'Request body XML parsing error.'
|
||||
return Response(response_xml_template.format(**failure_values), content_type="application/xml")
|
||||
|
||||
# Verify OAuth signing.
|
||||
@@ -498,10 +496,15 @@ oauth_consumer_key="", oauth_signature="frVp4JuvT1mVXlxktiAUjQ7%2F1cw%3D"'}
|
||||
self.verify_oauth_body_sign(request)
|
||||
except (ValueError, LTIError):
|
||||
failure_values['imsx_messageIdentifier'] = escape(imsx_messageIdentifier)
|
||||
failure_values['imsx_description'] = 'OAuth verification error.'
|
||||
return Response(response_xml_template.format(**failure_values), content_type="application/xml")
|
||||
|
||||
|
||||
real_user = self.system.get_real_user(urllib.unquote(sourcedId.split(':')[-1]))
|
||||
if not real_user: # that means we can't save to database, as we do not have real user id.
|
||||
failure_values['imsx_messageIdentifier'] = escape(imsx_messageIdentifier)
|
||||
failure_values['imsx_description'] = 'User not found.'
|
||||
return Response(response_xml_template.format(**failure_values), content_type="application/xml")
|
||||
|
||||
if action == 'replaceResultRequest':
|
||||
self.system.publish(
|
||||
event={
|
||||
@@ -518,9 +521,11 @@ oauth_consumer_key="", oauth_signature="frVp4JuvT1mVXlxktiAUjQ7%2F1cw%3D"'}
|
||||
'imsx_messageIdentifier': escape(imsx_messageIdentifier),
|
||||
'response': '<replaceResultResponse/>'
|
||||
}
|
||||
log.debug("[LTI]: Grade is saved.")
|
||||
return Response(response_xml_template.format(**values), content_type="application/xml")
|
||||
|
||||
unsupported_values['imsx_messageIdentifier'] = escape(imsx_messageIdentifier)
|
||||
log.debug("[LTI]: Incorrect action.")
|
||||
return Response(response_xml_template.format(**unsupported_values), content_type='application/xml')
|
||||
|
||||
|
||||
@@ -549,6 +554,7 @@ oauth_consumer_key="", oauth_signature="frVp4JuvT1mVXlxktiAUjQ7%2F1cw%3D"'}
|
||||
# Raise exception if score is not float or not in range 0.0-1.0 regarding spec.
|
||||
score = float(score)
|
||||
if not 0 <= score <= 1:
|
||||
log.debug("[LTI]: Score not in range.")
|
||||
raise LTIError
|
||||
|
||||
return imsx_messageIdentifier, sourcedId, score, action
|
||||
@@ -578,7 +584,7 @@ oauth_consumer_key="", oauth_signature="frVp4JuvT1mVXlxktiAUjQ7%2F1cw%3D"'}
|
||||
|
||||
sha1 = hashlib.sha1()
|
||||
sha1.update(request.body)
|
||||
oauth_body_hash = base64.b64encode(sha1.hexdigest())
|
||||
oauth_body_hash = base64.b64encode(sha1.digest())
|
||||
|
||||
oauth_params = signature.collect_parameters(headers=headers, exclude_oauth_signature=False)
|
||||
oauth_headers =dict(oauth_params)
|
||||
@@ -590,8 +596,11 @@ oauth_consumer_key="", oauth_signature="frVp4JuvT1mVXlxktiAUjQ7%2F1cw%3D"'}
|
||||
params=oauth_headers.items(),
|
||||
signature=oauth_signature
|
||||
)
|
||||
if (oauth_body_hash != oauth_headers.get('oauth_body_hash') or
|
||||
not signature.verify_hmac_sha1(mock_request, client_secret)):
|
||||
if oauth_body_hash != oauth_headers.get('oauth_body_hash'):
|
||||
log.debug("[LTI]: OAuth body hash verification is failed.")
|
||||
raise LTIError
|
||||
if not signature.verify_hmac_sha1(mock_request, client_secret):
|
||||
log.debug("[LTI]: OAuth signature verification is failed.")
|
||||
raise LTIError
|
||||
|
||||
def get_client_key_secret(self):
|
||||
|
||||
@@ -2,14 +2,21 @@
|
||||
"""Test for LTI Xmodule functional logic."""
|
||||
|
||||
from mock import Mock, patch, PropertyMock
|
||||
import mock
|
||||
import textwrap
|
||||
import json
|
||||
from lxml import etree
|
||||
import json
|
||||
from webob.request import Request
|
||||
from copy import copy
|
||||
from collections import OrderedDict
|
||||
import urllib
|
||||
import oauthlib
|
||||
import hashlib
|
||||
import base64
|
||||
|
||||
from xmodule.lti_module import LTIDescriptor
|
||||
|
||||
from xmodule.lti_module import LTIDescriptor, LTIError
|
||||
|
||||
from . import LogicTest
|
||||
|
||||
@@ -48,7 +55,6 @@ class LTIModuleTest(LogicTest):
|
||||
</imsx_POXEnvelopeRequest>
|
||||
""")
|
||||
self.system.get_real_user = Mock()
|
||||
self.xmodule.get_client_key_secret = Mock(return_value=('key', 'secret'))
|
||||
self.system.publish = Mock()
|
||||
|
||||
self.user_id = self.xmodule.runtime.anonymous_student_id
|
||||
@@ -96,6 +102,7 @@ class LTIModuleTest(LogicTest):
|
||||
def test_authorization_header_not_present(self):
|
||||
"""
|
||||
Request has no Authorization header.
|
||||
|
||||
This is an unknown service request, i.e., it is not a part of the original service specification.
|
||||
"""
|
||||
request = Request(self.environ)
|
||||
@@ -105,7 +112,7 @@ class LTIModuleTest(LogicTest):
|
||||
expected_response = {
|
||||
'action': None,
|
||||
'code_major': 'failure',
|
||||
'description': 'The request has failed.',
|
||||
'description': 'OAuth verification error.',
|
||||
'messageIdentifier': self.DEFAULTS['messageIdentifier'],
|
||||
}
|
||||
|
||||
@@ -115,6 +122,7 @@ class LTIModuleTest(LogicTest):
|
||||
def test_authorization_header_empty(self):
|
||||
"""
|
||||
Request Authorization header has no value.
|
||||
|
||||
This is an unknown service request, i.e., it is not a part of the original service specification.
|
||||
"""
|
||||
request = Request(self.environ)
|
||||
@@ -125,10 +133,29 @@ class LTIModuleTest(LogicTest):
|
||||
expected_response = {
|
||||
'action': None,
|
||||
'code_major': 'failure',
|
||||
'description': 'The request has failed.',
|
||||
'description': 'OAuth verification error.',
|
||||
'messageIdentifier': self.DEFAULTS['messageIdentifier'],
|
||||
}
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertDictEqual(expected_response, real_response)
|
||||
|
||||
def test_real_user_is_none(self):
|
||||
"""
|
||||
If we have no real user, we should send back failure response.
|
||||
"""
|
||||
self.xmodule.verify_oauth_body_sign = Mock()
|
||||
self.xmodule.has_score = True
|
||||
self.system.get_real_user = Mock(return_value=None)
|
||||
request = Request(self.environ)
|
||||
request.body = self.get_request_body()
|
||||
response = self.xmodule.grade_handler(request, '')
|
||||
real_response = self.get_response_values(response)
|
||||
expected_response = {
|
||||
'action': None,
|
||||
'code_major': 'failure',
|
||||
'description': 'User not found.',
|
||||
'messageIdentifier': self.DEFAULTS['messageIdentifier'],
|
||||
}
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertDictEqual(expected_response, real_response)
|
||||
|
||||
@@ -144,10 +171,9 @@ class LTIModuleTest(LogicTest):
|
||||
expected_response = {
|
||||
'action': None,
|
||||
'code_major': 'failure',
|
||||
'description': 'The request has failed.',
|
||||
'description': 'Request body XML parsing error.',
|
||||
'messageIdentifier': 'unknown',
|
||||
}
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertDictEqual(expected_response, real_response)
|
||||
|
||||
@@ -163,10 +189,9 @@ class LTIModuleTest(LogicTest):
|
||||
expected_response = {
|
||||
'action': None,
|
||||
'code_major': 'failure',
|
||||
'description': 'The request has failed.',
|
||||
'description': 'Request body XML parsing error.',
|
||||
'messageIdentifier': 'unknown',
|
||||
}
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertDictEqual(expected_response, real_response)
|
||||
|
||||
@@ -186,7 +211,6 @@ class LTIModuleTest(LogicTest):
|
||||
'description': 'Target does not support the requested operation.',
|
||||
'messageIdentifier': self.DEFAULTS['messageIdentifier'],
|
||||
}
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertDictEqual(expected_response, real_response)
|
||||
|
||||
@@ -199,7 +223,6 @@ class LTIModuleTest(LogicTest):
|
||||
request = Request(self.environ)
|
||||
request.body = self.get_request_body()
|
||||
response = self.xmodule.grade_handler(request, '')
|
||||
code_major, description, messageIdentifier, action = self.get_response_values(response)
|
||||
description_expected = 'Score for {sourcedId} is now {score}'.format(
|
||||
sourcedId=self.DEFAULTS['sourcedId'],
|
||||
score=self.DEFAULTS['grade'],
|
||||
@@ -221,20 +244,13 @@ class LTIModuleTest(LogicTest):
|
||||
self.assertEqual(real_user_id, expected_user_id)
|
||||
|
||||
def test_outcome_service_url(self):
|
||||
expected_outcome_service_url = 'http://{host}{path}'.format(
|
||||
expected_outcome_service_url = 'https://{host}{path}'.format(
|
||||
host=self.xmodule.runtime.hostname,
|
||||
path=self.xmodule.runtime.handler_url(self.xmodule, 'grade_handler', thirdparty=True).rstrip('/?')
|
||||
)
|
||||
|
||||
real_outcome_service_url = self.xmodule.get_outcome_service_url()
|
||||
self.assertEqual(real_outcome_service_url, expected_outcome_service_url)
|
||||
|
||||
def test_get_form_path(self):
|
||||
expected_form_path = self.xmodule.runtime.handler_url(self.xmodule, 'preview_handler').rstrip('/?')
|
||||
|
||||
real_form_path = self.xmodule.get_form_path()
|
||||
self.assertEqual(real_form_path, expected_form_path)
|
||||
|
||||
def test_resource_link_id(self):
|
||||
with patch('xmodule.lti_module.LTIModule.id', new_callable=PropertyMock) as mock_id:
|
||||
mock_id.return_value = self.module_id
|
||||
@@ -242,7 +258,6 @@ class LTIModuleTest(LogicTest):
|
||||
real_resource_link_id = self.xmodule.get_resource_link_id()
|
||||
self.assertEqual(real_resource_link_id, expected_resource_link_id)
|
||||
|
||||
|
||||
def test_lis_result_sourcedid(self):
|
||||
with patch('xmodule.lti_module.LTIModule.id', new_callable=PropertyMock) as mock_id:
|
||||
mock_id.return_value = self.module_id
|
||||
@@ -251,11 +266,127 @@ class LTIModuleTest(LogicTest):
|
||||
self.assertEqual(real_lis_result_sourcedid, expected_sourcedId)
|
||||
|
||||
|
||||
def test_verify_oauth_body_sign(self):
|
||||
pass
|
||||
@patch('xmodule.course_module.CourseDescriptor.id_to_location')
|
||||
def test_client_key_secret(self, test):
|
||||
"""
|
||||
LTI module gets client key and secret provided.
|
||||
"""
|
||||
#this adds lti passports to system
|
||||
mocked_course = Mock(lti_passports = ['lti_id:test_client:test_secret'])
|
||||
modulestore = Mock()
|
||||
modulestore.get_item.return_value = mocked_course
|
||||
runtime = Mock(modulestore=modulestore)
|
||||
self.xmodule.descriptor.runtime = runtime
|
||||
self.xmodule.lti_id = "lti_id"
|
||||
key, secret = self.xmodule.get_client_key_secret()
|
||||
expected = ('test_client', 'test_secret')
|
||||
self.assertEqual(expected, (key, secret))
|
||||
|
||||
def test_client_key_secret(self):
|
||||
pass
|
||||
@patch('xmodule.course_module.CourseDescriptor.id_to_location')
|
||||
def test_client_key_secret_not_provided(self, test):
|
||||
"""
|
||||
LTI module attempts to get client key and secret provided in cms.
|
||||
|
||||
There are key and secret but not for specific LTI.
|
||||
"""
|
||||
|
||||
#this adds lti passports to system
|
||||
mocked_course = Mock(lti_passports = ['test_id:test_client:test_secret'])
|
||||
modulestore = Mock()
|
||||
modulestore.get_item.return_value = mocked_course
|
||||
runtime = Mock(modulestore=modulestore)
|
||||
self.xmodule.descriptor.runtime = runtime
|
||||
#set another lti_id
|
||||
self.xmodule.lti_id = "another_lti_id"
|
||||
key_secret = self.xmodule.get_client_key_secret()
|
||||
expected = ('','')
|
||||
self.assertEqual(expected, key_secret)
|
||||
|
||||
@patch('xmodule.course_module.CourseDescriptor.id_to_location')
|
||||
def test_bad_client_key_secret(self, test):
|
||||
"""
|
||||
LTI module attempts to get client key and secret provided in cms.
|
||||
|
||||
There are key and secret provided in wrong format.
|
||||
"""
|
||||
#this adds lti passports to system
|
||||
mocked_course = Mock(lti_passports = ['test_id_test_client_test_secret'])
|
||||
modulestore = Mock()
|
||||
modulestore.get_item.return_value = mocked_course
|
||||
runtime = Mock(modulestore=modulestore)
|
||||
self.xmodule.descriptor.runtime = runtime
|
||||
self.xmodule.lti_id = 'lti_id'
|
||||
with self.assertRaises(LTIError):
|
||||
self.xmodule.get_client_key_secret()
|
||||
|
||||
@patch('xmodule.lti_module.signature.verify_hmac_sha1', return_value=True)
|
||||
@patch('xmodule.lti_module.LTIModule.get_client_key_secret', return_value=('test_client_key', u'test_client_secret'))
|
||||
def test_successful_verify_oauth_body_sign(self, get_key_secret, mocked_verify):
|
||||
"""
|
||||
Test if OAuth signing was successful.
|
||||
"""
|
||||
try:
|
||||
self.xmodule.verify_oauth_body_sign(self.get_signed_grade_mock_request())
|
||||
except LTIError:
|
||||
self.fail("verify_oauth_body_sign() raised LTIError unexpectedly!")
|
||||
|
||||
@patch('xmodule.lti_module.signature.verify_hmac_sha1', return_value=False)
|
||||
@patch('xmodule.lti_module.LTIModule.get_client_key_secret', return_value=('test_client_key', u'test_client_secret'))
|
||||
def test_failed_verify_oauth_body_sign(self, get_key_secret, mocked_verify):
|
||||
"""
|
||||
Oauth signing verify fail.
|
||||
"""
|
||||
with self.assertRaises(LTIError):
|
||||
req = self.get_signed_grade_mock_request()
|
||||
self.xmodule.verify_oauth_body_sign(req)
|
||||
|
||||
def get_signed_grade_mock_request(self):
|
||||
"""
|
||||
Example of signed request from LTI Provider.
|
||||
"""
|
||||
mock_request = Mock()
|
||||
mock_request.headers = {
|
||||
'X-Requested-With': 'XMLHttpRequest',
|
||||
'Content-Type': 'application/xml',
|
||||
'Authorization': u'OAuth oauth_nonce="135685044251684026041377608307", \
|
||||
oauth_timestamp="1234567890", oauth_version="1.0", \
|
||||
oauth_signature_method="HMAC-SHA1", \
|
||||
oauth_consumer_key="test_client_key", \
|
||||
oauth_signature="my_signature%3D", \
|
||||
oauth_body_hash="gz+PeJZuF2//n9hNUnDj2v5kN70="'
|
||||
}
|
||||
mock_request.url = u'http://testurl'
|
||||
mock_request.http_method = u'POST'
|
||||
mock_request.body = textwrap.dedent("""
|
||||
<?xml version = "1.0" encoding = "UTF-8"?>
|
||||
<imsx_POXEnvelopeRequest xmlns="http://www.imsglobal.org/services/ltiv1p1/xsd/imsoms_v1p0">
|
||||
</imsx_POXEnvelopeRequest>
|
||||
""")
|
||||
return mock_request
|
||||
|
||||
def test_good_custom_params(self):
|
||||
"""
|
||||
Custom parameters are presented in right format.
|
||||
"""
|
||||
self.xmodule.custom_parameters = ['test_custom_params=test_custom_param_value']
|
||||
self.xmodule.get_client_key_secret = Mock(return_value=('test_client_key', 'test_client_secret'))
|
||||
self.xmodule.oauth_params = Mock()
|
||||
self.xmodule.get_input_fields()
|
||||
self.xmodule.oauth_params.assert_called_with(
|
||||
{u'custom_test_custom_params': u'test_custom_param_value'},
|
||||
'test_client_key', 'test_client_secret'
|
||||
)
|
||||
|
||||
def test_bad_custom_params(self):
|
||||
"""
|
||||
Custom parameters are presented in wrong format.
|
||||
"""
|
||||
bad_custom_params = ['test_custom_params: test_custom_param_value']
|
||||
self.xmodule.custom_parameters = bad_custom_params
|
||||
self.xmodule.get_client_key_secret = Mock(return_value=('test_client_key', 'test_client_secret'))
|
||||
self.xmodule.oauth_params = Mock()
|
||||
with self.assertRaises(LTIError):
|
||||
self.xmodule.get_input_fields()
|
||||
|
||||
def test_max_score(self):
|
||||
self.xmodule.weight = 100.0
|
||||
|
||||
@@ -1016,6 +1016,9 @@ class ModuleSystem(ConfigurableFragmentWrapper, Runtime): # pylint: disable=abs
|
||||
|
||||
error_descriptor_class - The class to use to render XModules with errors
|
||||
|
||||
get_real_user - function that takes `anonymous_student_id` and returns real user_id,
|
||||
associated with `anonymous_student_id`.
|
||||
|
||||
"""
|
||||
|
||||
# Right now, usage_store is unused, and field_data is always supplanted
|
||||
|
||||
@@ -38,6 +38,10 @@ def setup_mock_lti_server():
|
||||
'lti_endpoint': 'correct_lti_endpoint'
|
||||
}
|
||||
|
||||
# Flag for acceptance tests used for creating right callback_url and sending
|
||||
# graded result. Used in MockLTIRequestHandler.
|
||||
server.test_mode = True
|
||||
|
||||
# Store the server instance in lettuce's world
|
||||
# so that other steps can access it
|
||||
# (and we can shut it down later)
|
||||
|
||||
@@ -1,3 +1,13 @@
|
||||
"""
|
||||
LTI Server
|
||||
|
||||
What is supported:
|
||||
------------------
|
||||
|
||||
1.) This LTI Provider can service only one Tool Consumer at the same time. It is
|
||||
not possible to have this LTI multiple times on a single page in LMS.
|
||||
|
||||
"""
|
||||
from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler
|
||||
from uuid import uuid4
|
||||
import textwrap
|
||||
@@ -35,88 +45,48 @@ class MockLTIRequestHandler(BaseHTTPRequestHandler):
|
||||
def do_GET(self):
|
||||
'''
|
||||
Handle a GET request from the client and sends response back.
|
||||
'''
|
||||
|
||||
Used for checking LTI Provider started correctly.
|
||||
'''
|
||||
self.send_response(200, 'OK')
|
||||
self.send_header('Content-type', 'html')
|
||||
self.end_headers()
|
||||
|
||||
response_str = """<html><head><title>TEST TITLE</title></head>
|
||||
<body>I have stored grades.</body></html>"""
|
||||
|
||||
<body>This is LTI Provider.</body></html>"""
|
||||
self.wfile.write(response_str)
|
||||
|
||||
self._send_graded_result()
|
||||
|
||||
|
||||
|
||||
def do_POST(self):
|
||||
'''
|
||||
Handle a POST request from the client and sends response back.
|
||||
'''
|
||||
|
||||
'''
|
||||
logger.debug("LTI provider received POST request {} to path {}".format(
|
||||
str(self.post_dict),
|
||||
self.path)
|
||||
) # Log the request
|
||||
'''
|
||||
# Respond to grade request
|
||||
if 'grade' in self.path and self._send_graded_result().status_code == 200:
|
||||
status_message = 'LTI consumer (edX) responded with XML content:<br>' + self.server.grade_data['TC answer']
|
||||
self.server.grade_data['callback_url'] = None
|
||||
self._send_response(status_message, 200)
|
||||
# Respond to request with correct lti endpoint:
|
||||
elif self._is_correct_lti_request():
|
||||
self.post_dict = self._post_dict()
|
||||
correct_keys = [
|
||||
'user_id',
|
||||
'role',
|
||||
'oauth_nonce',
|
||||
'oauth_timestamp',
|
||||
'oauth_consumer_key',
|
||||
'lti_version',
|
||||
'oauth_signature_method',
|
||||
'oauth_version',
|
||||
'oauth_signature',
|
||||
'lti_message_type',
|
||||
'oauth_callback',
|
||||
'lis_outcome_service_url',
|
||||
'lis_result_sourcedid',
|
||||
'launch_presentation_return_url',
|
||||
# 'lis_person_sourcedid', optional, not used now.
|
||||
'resource_link_id',
|
||||
]
|
||||
if sorted(correct_keys) != sorted(self.post_dict.keys()):
|
||||
status_message = "Incorrect LTI header"
|
||||
params = {k: v for k, v in self.post_dict.items() if k != 'oauth_signature'}
|
||||
if self.server.check_oauth_signature(params, self.post_dict.get('oauth_signature', "")):
|
||||
status_message = "This is LTI tool. Success."
|
||||
# set data for grades what need to be stored as server data
|
||||
if 'lis_outcome_service_url' in self.post_dict:
|
||||
self.server.grade_data = {
|
||||
'callback_url': self.post_dict.get('lis_outcome_service_url'),
|
||||
'sourcedId': self.post_dict.get('lis_result_sourcedid')
|
||||
}
|
||||
else:
|
||||
params = {k: v for k, v in self.post_dict.items() if k != 'oauth_signature'}
|
||||
if self.server.check_oauth_signature(params, self.post_dict['oauth_signature']):
|
||||
status_message = "This is LTI tool. Success."
|
||||
else:
|
||||
status_message = "Wrong LTI signature"
|
||||
# set data for grades
|
||||
# what need to be stored as server data
|
||||
self.server.grade_data = {
|
||||
'callback_url': self.post_dict["lis_outcome_service_url"],
|
||||
'sourcedId': self.post_dict['lis_result_sourcedid']
|
||||
}
|
||||
status_message = "Wrong LTI signature"
|
||||
self._send_response(status_message, 200)
|
||||
else:
|
||||
status_message = "Invalid request URL"
|
||||
self._send_response(status_message, 500)
|
||||
|
||||
self._send_head()
|
||||
self._send_response(status_message)
|
||||
|
||||
def _send_head(self):
|
||||
def _send_head(self, status_code):
|
||||
'''
|
||||
Send the response code and MIME headers
|
||||
'''
|
||||
self.send_response(200)
|
||||
'''
|
||||
if self._is_correct_lti_request():
|
||||
self.send_response(200)
|
||||
else:
|
||||
self.send_response(500)
|
||||
'''
|
||||
self.send_response(status_code)
|
||||
self.send_header('Content-type', 'text/html')
|
||||
self.end_headers()
|
||||
|
||||
@@ -144,17 +114,17 @@ class MockLTIRequestHandler(BaseHTTPRequestHandler):
|
||||
self.server.cookie = {}
|
||||
referer = urlparse.urlparse(self.headers.getheader('referer'))
|
||||
self.server.referer_host = "{}://{}".format(referer.scheme, referer.netloc)
|
||||
self.server.referer_netloc = referer.netloc
|
||||
return post_dict
|
||||
|
||||
def _send_graded_result(self):
|
||||
|
||||
"""
|
||||
Send grade request.
|
||||
"""
|
||||
values = {
|
||||
'textString': 0.5,
|
||||
'sourcedId': self.server.grade_data['sourcedId'],
|
||||
'imsx_messageIdentifier': uuid4().hex,
|
||||
}
|
||||
|
||||
payload = textwrap.dedent("""
|
||||
<?xml version = "1.0" encoding = "UTF-8"?>
|
||||
<imsx_POXEnvelopeRequest xmlns="http://www.imsglobal.org/services/ltiv1p1/xsd/imsoms_v1p0">
|
||||
@@ -182,15 +152,22 @@ class MockLTIRequestHandler(BaseHTTPRequestHandler):
|
||||
</imsx_POXEnvelopeRequest>
|
||||
""")
|
||||
data = payload.format(**values)
|
||||
# temporarily changed to get for easy view in browser
|
||||
# get relative part, because host name is different in a) manual tests b) acceptance tests c) demos
|
||||
relative_url = urlparse.urlparse(self.server.grade_data['callback_url']).path
|
||||
url = self.server.referer_host + relative_url
|
||||
if getattr(self.server, 'test_mode', None):
|
||||
relative_url = urlparse.urlparse(self.server.grade_data['callback_url']).path
|
||||
url = self.server.referer_host + relative_url
|
||||
else:
|
||||
url = self.server.grade_data['callback_url']
|
||||
|
||||
headers = {'Content-Type': 'application/xml', 'X-Requested-With': 'XMLHttpRequest'}
|
||||
|
||||
headers['Authorization'] = self.oauth_sign(url, data)
|
||||
|
||||
# We can't mock requests in unit tests, because we use them, but we need
|
||||
# them to be mocked only for this one case.
|
||||
if getattr(self.server, 'run_inside_unittest_flag', None):
|
||||
response = mock.Mock(status_code=200, url=url, data=data, headers=headers)
|
||||
return response
|
||||
|
||||
response = requests.post(
|
||||
url,
|
||||
data=data,
|
||||
@@ -199,45 +176,58 @@ class MockLTIRequestHandler(BaseHTTPRequestHandler):
|
||||
self.server.grade_data['TC answer'] = response.content
|
||||
return response
|
||||
|
||||
def _send_response(self, message):
|
||||
def _send_response(self, message, status_code):
|
||||
'''
|
||||
Send message back to the client
|
||||
'''
|
||||
self._send_head(status_code)
|
||||
if getattr(self.server, 'grade_data', False): # lti can be graded
|
||||
response_str = textwrap.dedent("""
|
||||
<html>
|
||||
<head>
|
||||
<title>TEST TITLE</title>
|
||||
</head>
|
||||
<body>
|
||||
<div>
|
||||
<h2>Graded IFrame loaded</h2>
|
||||
<h3>Server response is:</h3>
|
||||
<h3 class="result">{}</h3>
|
||||
</div>
|
||||
<form action="{url}/grade" method="post">
|
||||
<input type="submit" name="submit-button" value="Submit">
|
||||
</form>
|
||||
</body>
|
||||
</html>
|
||||
""").format(message, url="http://%s:%s" % self.server.server_address)
|
||||
else: # lti can't be graded
|
||||
response_str = textwrap.dedent("""
|
||||
<html>
|
||||
<head>
|
||||
<title>TEST TITLE</title>
|
||||
</head>
|
||||
<body>
|
||||
<div>
|
||||
<h2>IFrame loaded</h2>
|
||||
<h3>Server response is:</h3>
|
||||
<h3 class="result">{}</h3>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
""").format(message)
|
||||
|
||||
if self.server.grade_data['callback_url']:
|
||||
response_str = """<html><head><title>TEST TITLE</title></head>
|
||||
<body>
|
||||
<div><h2>Graded IFrame loaded</h2> \
|
||||
<h3>Server response is:</h3>\
|
||||
<h3 class="result">{}</h3></div>
|
||||
<form action="{url}/grade" method="post">
|
||||
<input type="submit" name="submit-button" value="Submit">
|
||||
</form>
|
||||
|
||||
</body></html>""".format(message, url="http://%s:%s" % self.server.server_address)
|
||||
else:
|
||||
response_str = """<html><head><title>TEST TITLE</title></head>
|
||||
<body>
|
||||
<div><h2>IFrame loaded</h2> \
|
||||
<h3>Server response is:</h3>\
|
||||
<h3 class="result">{}</h3></div>
|
||||
</body></html>""".format(message)
|
||||
|
||||
# Log the response
|
||||
logger.debug("LTI: sent response {}".format(response_str))
|
||||
|
||||
self.wfile.write(response_str)
|
||||
|
||||
def _is_correct_lti_request(self):
|
||||
'''If url to LTI tool is correct.'''
|
||||
'''
|
||||
If url to LTI tool is correct.
|
||||
'''
|
||||
return self.server.oauth_settings['lti_endpoint'] in self.path
|
||||
|
||||
def oauth_sign(self, url, body):
|
||||
"""
|
||||
Signs request and returns signed body and headers.
|
||||
|
||||
"""
|
||||
|
||||
client = oauthlib.oauth1.Client(
|
||||
client_key=unicode(self.server.oauth_settings['client_key']),
|
||||
client_secret=unicode(self.server.oauth_settings['client_secret'])
|
||||
@@ -250,7 +240,7 @@ class MockLTIRequestHandler(BaseHTTPRequestHandler):
|
||||
#Calculate and encode body hash. See http://oauth.googlecode.com/svn/spec/ext/body_hash/1.0/oauth-bodyhash.html
|
||||
sha1 = hashlib.sha1()
|
||||
sha1.update(body)
|
||||
oauth_body_hash = base64.b64encode(sha1.hexdigest())
|
||||
oauth_body_hash = base64.b64encode(sha1.digest())
|
||||
__, headers, __ = client.sign(
|
||||
unicode(url.strip()),
|
||||
http_method=u'POST',
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
"""
|
||||
Mock LTI server for manual testing.
|
||||
|
||||
Used for manual testing and testing on sandbox.
|
||||
"""
|
||||
|
||||
import threading
|
||||
@@ -18,6 +20,10 @@ server.oauth_settings = {
|
||||
}
|
||||
server.server_host = server_host
|
||||
|
||||
# If in test mode mock lti server will make callback url using referer host.
|
||||
# Used in MockLTIRequestHandler when sending graded result.
|
||||
server.test_mode = True
|
||||
|
||||
try:
|
||||
server.serve_forever()
|
||||
except KeyboardInterrupt:
|
||||
|
||||
@@ -11,7 +11,6 @@ import requests
|
||||
from mock_lti_server import MockLTIServer
|
||||
|
||||
|
||||
|
||||
class MockLTIServerTest(unittest.TestCase):
|
||||
'''
|
||||
A mock version of the LTI provider server that listens on a local
|
||||
@@ -33,6 +32,10 @@ class MockLTIServerTest(unittest.TestCase):
|
||||
'lti_base': 'http://{}:{}/'.format(server_host, server_port),
|
||||
'lti_endpoint': 'correct_lti_endpoint'
|
||||
}
|
||||
self.server.run_inside_unittest_flag = True
|
||||
#flag for creating right callback_url
|
||||
self.server.test_mode = True
|
||||
|
||||
# Start the server in a separate daemon thread
|
||||
server_thread = threading.Thread(target=self.server.serve_forever)
|
||||
server_thread.daemon = True
|
||||
@@ -43,6 +46,23 @@ class MockLTIServerTest(unittest.TestCase):
|
||||
# Stop the server, freeing up the port
|
||||
self.server.shutdown()
|
||||
|
||||
|
||||
def test_wrong_header(self):
|
||||
"""
|
||||
Tests that LTI server processes request with right program path but with wrong header.
|
||||
"""
|
||||
#wrong number of params and no signature
|
||||
payload = {
|
||||
'user_id': 'default_user_id',
|
||||
'role': 'student',
|
||||
'oauth_nonce': '',
|
||||
'oauth_timestamp': '',
|
||||
}
|
||||
uri = self.server.oauth_settings['lti_base'] + self.server.oauth_settings['lti_endpoint']
|
||||
headers = {'referer': 'http://localhost:8000/'}
|
||||
response = requests.post(uri, data=payload, headers=headers)
|
||||
self.assertIn('Wrong LTI signature', response.content)
|
||||
|
||||
def test_wrong_signature(self):
|
||||
"""
|
||||
Tests that LTI server processes request with right program
|
||||
@@ -53,7 +73,7 @@ class MockLTIServerTest(unittest.TestCase):
|
||||
'role': 'student',
|
||||
'oauth_nonce': '',
|
||||
'oauth_timestamp': '',
|
||||
'oauth_consumer_key': 'client_key',
|
||||
'oauth_consumer_key': 'test_client_key',
|
||||
'lti_version': 'LTI-1p0',
|
||||
'oauth_signature_method': 'HMAC-SHA1',
|
||||
'oauth_version': '1.0',
|
||||
@@ -65,25 +85,48 @@ class MockLTIServerTest(unittest.TestCase):
|
||||
'lis_result_sourcedid': '',
|
||||
'resource_link_id':'',
|
||||
}
|
||||
|
||||
uri = self.server.oauth_settings['lti_base'] + self.server.oauth_settings['lti_endpoint']
|
||||
headers = {'referer': 'http://localhost:8000/'}
|
||||
response = requests.post(uri, data=payload, headers=headers)
|
||||
self.assertIn('Wrong LTI signature', response.content)
|
||||
|
||||
self.assertTrue('Wrong LTI signature' in response.content)
|
||||
|
||||
|
||||
def test_success_response_launch_lti(self):
|
||||
"""
|
||||
Success lti launch.
|
||||
"""
|
||||
payload = {
|
||||
'user_id': 'default_user_id',
|
||||
'role': 'student',
|
||||
'oauth_nonce': '',
|
||||
'oauth_timestamp': '',
|
||||
'oauth_consumer_key': 'test_client_key',
|
||||
'lti_version': 'LTI-1p0',
|
||||
'oauth_signature_method': 'HMAC-SHA1',
|
||||
'oauth_version': '1.0',
|
||||
'oauth_signature': '',
|
||||
'lti_message_type': 'basic-lti-launch-request',
|
||||
'oauth_callback': 'about:blank',
|
||||
'launch_presentation_return_url': '',
|
||||
'lis_outcome_service_url': '',
|
||||
'lis_result_sourcedid': '',
|
||||
'resource_link_id':'',
|
||||
}
|
||||
self.server.check_oauth_signature = Mock(return_value=True)
|
||||
|
||||
uri = self.server.oauth_settings['lti_base'] + self.server.oauth_settings['lti_endpoint']
|
||||
headers = {'referer': 'http://localhost:8000/'}
|
||||
response = requests.post(uri, data=payload, headers=headers)
|
||||
self.assertIn('This is LTI tool. Success.', response.content)
|
||||
|
||||
def test_send_graded_result(self):
|
||||
|
||||
payload = {
|
||||
'user_id': 'default_user_id',
|
||||
'role': 'student',
|
||||
'oauth_nonce': '',
|
||||
'oauth_timestamp': '',
|
||||
'oauth_consumer_key': 'client_key',
|
||||
'oauth_consumer_key': 'test_client_key',
|
||||
'lti_version': 'LTI-1p0',
|
||||
'oauth_signature_method': 'HMAC-SHA1',
|
||||
'oauth_version': '1.0',
|
||||
@@ -94,14 +137,18 @@ class MockLTIServerTest(unittest.TestCase):
|
||||
'lis_outcome_service_url': '',
|
||||
'lis_result_sourcedid': '',
|
||||
'resource_link_id':'',
|
||||
"lis_outcome_service_url": '',
|
||||
}
|
||||
self.server.check_oauth_signature = Mock(return_value=True)
|
||||
|
||||
uri = self.server.oauth_settings['lti_base'] + self.server.oauth_settings['lti_endpoint']
|
||||
headers = {'referer': 'http://localhost:8000/'}
|
||||
|
||||
response = requests.post(uri, data=payload, headers=headers)
|
||||
|
||||
self.assertTrue('This is LTI tool. Success.' in response.content)
|
||||
|
||||
uri = self.server.oauth_settings['lti_base'] + self.server.oauth_settings['lti_endpoint']
|
||||
#this is the uri for sending grade from lti
|
||||
headers = {'referer': 'http://localhost:8000/'}
|
||||
response = requests.post(uri, data=payload, headers=headers)
|
||||
self.assertIn('This is LTI tool. Success.', response.content)
|
||||
|
||||
self.server.grade_data['TC answer'] = "Test response"
|
||||
graded_response = requests.post('http://127.0.0.1:8034/grade')
|
||||
self.assertIn('Test response', graded_response.content)
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -33,7 +33,7 @@ class TestLTI(BaseTestXmodule):
|
||||
|
||||
sourcedId = u':'.join(urllib.quote(i) for i in (lti_id, module_id, user_id))
|
||||
|
||||
lis_outcome_service_url = 'http://{host}{path}'.format(
|
||||
lis_outcome_service_url = 'https://{host}{path}'.format(
|
||||
host=self.item_descriptor.xmodule_runtime.hostname,
|
||||
path=self.item_descriptor.xmodule_runtime.handler_url(self.item_module, 'grade_handler', thirdparty=True).rstrip('/?')
|
||||
)
|
||||
@@ -46,7 +46,6 @@ class TestLTI(BaseTestXmodule):
|
||||
u'role': u'student',
|
||||
|
||||
u'resource_link_id': module_id,
|
||||
u'lis_outcome_service_url': lis_outcome_service_url,
|
||||
u'lis_result_sourcedid': sourcedId,
|
||||
|
||||
u'oauth_nonce': mocked_nonce,
|
||||
@@ -59,6 +58,16 @@ class TestLTI(BaseTestXmodule):
|
||||
|
||||
saved_sign = oauthlib.oauth1.Client.sign
|
||||
|
||||
self.expected_context = {
|
||||
'display_name': self.item_module.display_name,
|
||||
'input_fields': self.correct_headers,
|
||||
'element_class': self.item_module.category,
|
||||
'element_id': self.item_module.location.html_id(),
|
||||
'launch_url': 'http://www.example.com', # default value
|
||||
'open_in_a_new_page': True,
|
||||
'form_url': self.item_descriptor.xmodule_runtime.handler_url(self.item_module, 'preview_handler').rstrip('/?'),
|
||||
}
|
||||
|
||||
def mocked_sign(self, *args, **kwargs):
|
||||
"""
|
||||
Mocked oauth1 sign function.
|
||||
@@ -79,21 +88,11 @@ class TestLTI(BaseTestXmodule):
|
||||
self.addCleanup(patcher.stop)
|
||||
|
||||
def test_lti_constructor(self):
|
||||
"""
|
||||
Makes sure that all parameters extracted.
|
||||
"""
|
||||
generated_context = self.item_module.render('student_view').content
|
||||
expected_context = {
|
||||
'display_name': self.item_module.display_name,
|
||||
'input_fields': self.correct_headers,
|
||||
'element_class': self.item_module.category,
|
||||
'element_id': self.item_module.location.html_id(),
|
||||
'launch_url': 'http://www.example.com', # default value
|
||||
'open_in_a_new_page': True,
|
||||
'form_url': self.item_descriptor.xmodule_runtime.handler_url(self.item_module, 'preview_handler').rstrip('/?'),
|
||||
}
|
||||
generated_content = self.item_module.render('student_view').content
|
||||
expected_content = self.runtime.render_template('lti.html', self.expected_context)
|
||||
self.assertEqual(generated_content, expected_content)
|
||||
|
||||
self.assertEqual(
|
||||
generated_context,
|
||||
self.runtime.render_template('lti.html', expected_context),
|
||||
)
|
||||
def test_lti_preview_handler(self):
|
||||
generated_content = self.item_module.preview_handler(None, None).body
|
||||
expected_content = self.runtime.render_template('lti_form.html', self.expected_context)
|
||||
self.assertEqual(generated_content, expected_content)
|
||||
|
||||
Reference in New Issue
Block a user