From ba7044e9a77605bc6cbe5c9c7d6e922863483176 Mon Sep 17 00:00:00 2001
From: Patrick Cockwell <patrick@opencraft.com>
Date: Thu, 2 Jul 2020 11:16:29 +0700
Subject: [PATCH] BD-02 BD-03 Add support for LTI1.1 embeds in course tabs and
elsewhere
---
Makefile | 2 +-
lti_consumer/exceptions.py | 4 +-
lti_consumer/lti.py | 306 --------------
lti_consumer/lti_1p1/README.rst | 71 ++++
lti_consumer/lti_1p1/__init__.py | 0
lti_consumer/lti_1p1/consumer.py | 390 ++++++++++++++++++
lti_consumer/lti_1p1/contrib/__init__.py | 0
lti_consumer/lti_1p1/contrib/django.py | 123 ++++++
.../lti_1p1/contrib/tests/__init__.py | 0
.../lti_1p1/contrib/tests/test_django.py | 136 ++++++
lti_consumer/lti_1p1/exceptions.py | 9 +
lti_consumer/{ => lti_1p1}/oauth.py | 40 +-
lti_consumer/lti_1p1/tests/__init__.py | 0
lti_consumer/lti_1p1/tests/test_consumer.py | 349 ++++++++++++++++
.../unit => lti_1p1/tests}/test_oauth.py | 16 +-
lti_consumer/lti_xblock.py | 220 +++++++---
lti_consumer/outcomes.py | 10 +-
lti_consumer/tests/unit/test_lti.py | 349 ----------------
lti_consumer/tests/unit/test_lti_consumer.py | 267 +++++++++---
lti_consumer/tests/unit/test_utils.py | 3 +-
requirements/travis.in | 2 -
setup.py | 2 +-
22 files changed, 1498 insertions(+), 801 deletions(-)
delete mode 100644 lti_consumer/lti.py
create mode 100644 lti_consumer/lti_1p1/README.rst
create mode 100644 lti_consumer/lti_1p1/__init__.py
create mode 100644 lti_consumer/lti_1p1/consumer.py
create mode 100644 lti_consumer/lti_1p1/contrib/__init__.py
create mode 100644 lti_consumer/lti_1p1/contrib/django.py
create mode 100644 lti_consumer/lti_1p1/contrib/tests/__init__.py
create mode 100644 lti_consumer/lti_1p1/contrib/tests/test_django.py
create mode 100644 lti_consumer/lti_1p1/exceptions.py
rename lti_consumer/{ => lti_1p1}/oauth.py (79%)
create mode 100644 lti_consumer/lti_1p1/tests/__init__.py
create mode 100644 lti_consumer/lti_1p1/tests/test_consumer.py
rename lti_consumer/{tests/unit => lti_1p1/tests}/test_oauth.py (88%)
delete mode 100644 lti_consumer/tests/unit/test_lti.py
diff --git a/Makefile b/Makefile
index 0d35045..de9cd7e 100644
--- a/Makefile
+++ b/Makefile
@@ -25,4 +25,4 @@ upgrade: ## update the requirements/*.txt files with the latest packages satisfy
# Let tox control the Django version version for tests
grep -e "^django==" requirements/test.txt > requirements/django.txt
sed '/^[dD]jango==/d' requirements/test.txt > requirements/test.tmp
- mv requirements/test.tmp requirements/test.txt
\ No newline at end of file
+ mv requirements/test.tmp requirements/test.txt
diff --git a/lti_consumer/exceptions.py b/lti_consumer/exceptions.py
index e80b55f..dec5469 100644
--- a/lti_consumer/exceptions.py
+++ b/lti_consumer/exceptions.py
@@ -1,9 +1,9 @@
"""
-Exceptions for the LTI Consumer XBlock.
+Exceptions for the LTI Consumer.
"""
class LtiError(Exception):
"""
- General error class for LTI XBlock.
+ General error class for LTI Consumer usage.
"""
diff --git a/lti_consumer/lti.py b/lti_consumer/lti.py
deleted file mode 100644
index 184b273..0000000
--- a/lti_consumer/lti.py
+++ /dev/null
@@ -1,306 +0,0 @@
-"""
-This module encapsulates code which implements the LTI specification.
-
-For more details see:
-https://www.imsglobal.org/activity/learning-tools-interoperability
-"""
-
-import json
-import logging
-
-import six.moves.urllib.error
-import six.moves.urllib.parse
-from six import text_type
-
-from .exceptions import LtiError
-from .oauth import get_oauth_request_signature, verify_oauth_body_signature
-
-log = logging.getLogger(__name__)
-
-
-def parse_result_json(json_str):
- """
- Helper method for verifying LTI 2.0 JSON object contained in the body of the request.
-
- The json_str must be loadable. It can either be an dict (object) or an array whose first element is an dict,
- in which case that first dict is considered.
- The dict must have the "@type" key with value equal to "Result",
- "resultScore" key with value equal to a number [0, 1], if "resultScore" is not
- included in the JSON body, score will be returned as None
- The "@context" key must be present, but we don't do anything with it. And the "comment" key may be
- present, in which case it must be a string.
-
- Arguments:
- json_str (unicode): The body of the LTI 2.0 results service request, which is a JSON string
-
- Returns:
- (float, str): (score, [optional]comment) if parsing is successful
-
- Raises:
- LtiError: if verification fails
- """
- try:
- json_obj = json.loads(json_str)
- except (ValueError, TypeError):
- msg = "Supplied JSON string in request body could not be decoded: {}".format(json_str)
- log.error("[LTI] %s", msg)
- raise LtiError(msg)
-
- # The JSON object must be a dict. If a non-empty list is passed in,
- # use the first element, but only if it is a dict
- if isinstance(json_obj, list) and len(json_obj) >= 1:
- json_obj = json_obj[0]
-
- if not isinstance(json_obj, dict):
- msg = ("Supplied JSON string is a list that does not contain an object as the first element. {}"
- .format(json_str))
- log.error("[LTI] %s", msg)
- raise LtiError(msg)
-
- # '@type' must be "Result"
- result_type = json_obj.get("@type")
- if result_type != "Result":
- msg = "JSON object does not contain correct @type attribute (should be 'Result', is z{})".format(result_type)
- log.error("[LTI] %s", msg)
- raise LtiError(msg)
-
- # '@context' must be present as a key
- if '@context' not in json_obj:
- msg = "JSON object does not contain required key @context"
- log.error("[LTI] %s", msg)
- raise LtiError(msg)
-
- # Return None if the resultScore key is missing, this condition
- # will be handled by the upstream caller of this function
- if "resultScore" not in json_obj:
- score = None
- else:
- # if present, 'resultScore' must be a number between 0 and 1 inclusive
- try:
- score = float(json_obj.get('resultScore', "unconvertable")) # Check if float is present and the right type
- if not 0.0 <= score <= 1.0:
- msg = 'score value outside the permitted range of 0.0-1.0.'
- log.error("[LTI] %s", msg)
- raise LtiError(msg)
- except (TypeError, ValueError) as err:
- msg = "Could not convert resultScore to float: {}".format(str(err))
- log.error("[LTI] %s", msg)
- raise LtiError(msg)
-
- return score, json_obj.get('comment', "")
-
-
-class LtiConsumer(object): # pylint: disable=bad-option-value, useless-object-inheritance
- """
- Limited implementation of the LTI 1.1/2.0 specification.
-
- For the LTI 1.1 specification see:
- https://www.imsglobal.org/specs/ltiv1p1
-
- For the LTI 2.0 specification see:
- https://www.imsglobal.org/specs/ltiv2p0
- """
- CONTENT_TYPE_RESULT_JSON = 'application/vnd.ims.lis.v2.result+json'
-
- def __init__(self, xblock):
- self.xblock = xblock
-
- def get_signed_lti_parameters(self):
- """
- Signs LTI launch request and returns signature and OAuth parameters.
-
- Arguments:
- None
-
- Returns:
- dict: LTI launch parameters
- """
-
- # Must have parameters for correct signing from LTI:
- lti_parameters = {
- text_type('user_id'): self.xblock.user_id,
- text_type('oauth_callback'): text_type('about:blank'),
- text_type('launch_presentation_return_url'): '',
- text_type('lti_message_type'): text_type('basic-lti-launch-request'),
- text_type('lti_version'): text_type('LTI-1p0'),
- text_type('roles'): self.xblock.role,
-
- # Parameters required for grading:
- text_type('resource_link_id'): self.xblock.resource_link_id,
- text_type('lis_result_sourcedid'): self.xblock.lis_result_sourcedid,
-
- text_type('context_id'): self.xblock.context_id,
- text_type('custom_component_display_name'): self.xblock.display_name,
-
- text_type('context_title'): self.xblock.course.display_name_with_default,
- text_type('context_label'): self.xblock.course.display_org_with_default,
- }
-
- if self.xblock.due:
- lti_parameters['custom_component_due_date'] = self.xblock.due.strftime('%Y-%m-%d %H:%M:%S')
- if self.xblock.graceperiod:
- lti_parameters['custom_component_graceperiod'] = str(self.xblock.graceperiod.total_seconds())
-
- if self.xblock.has_score:
- lti_parameters.update({
- text_type('lis_outcome_service_url'): self.xblock.outcome_service_url
- })
-
- self.xblock.user_email = ""
- self.xblock.user_username = ""
- self.xblock.user_language = ""
-
- # Username, email, and language can't be sent in studio mode, because the user object is not defined.
- # To test functionality test in LMS
-
- if callable(self.xblock.runtime.get_real_user):
- real_user_object = self.xblock.runtime.get_real_user(self.xblock.runtime.anonymous_student_id)
- self.xblock.user_email = getattr(real_user_object, "email", "")
- self.xblock.user_username = getattr(real_user_object, "username", "")
- user_preferences = getattr(real_user_object, "preferences", None)
-
- if user_preferences is not None:
- language_preference = user_preferences.filter(key='pref-lang')
- if len(language_preference) == 1:
- self.xblock.user_language = language_preference[0].value
-
- if self.xblock.ask_to_send_username and self.xblock.user_username:
- lti_parameters["lis_person_sourcedid"] = self.xblock.user_username
- if self.xblock.ask_to_send_email and self.xblock.user_email:
- lti_parameters["lis_person_contact_email_primary"] = self.xblock.user_email
- if self.xblock.user_language:
- lti_parameters["launch_presentation_locale"] = self.xblock.user_language
-
- # Appending custom parameter for signing.
- lti_parameters.update(self.xblock.prefixed_custom_parameters)
-
- for processor in self.xblock.get_parameter_processors():
- try:
- default_params = getattr(processor, 'lti_xblock_default_params', {})
- lti_parameters.update(default_params)
- lti_parameters.update(processor(self.xblock) or {})
- except Exception: # pylint: disable=broad-except
- # Log the error without causing a 500-error.
- # Useful for catching casual runtime errors in the processors.
- log.exception('Error in XBlock LTI parameter processor "%s"', processor)
-
- headers = {
- # This is needed for body encoding:
- 'Content-Type': 'application/x-www-form-urlencoded',
- }
-
- key, secret = self.xblock.lti_provider_key_secret
- oauth_signature = get_oauth_request_signature(key, secret, self.xblock.launch_url, headers, lti_parameters)
-
- # Parse headers to pass to template as part of context:
- oauth_signature = dict([param.strip().replace('"', '').split('=') for param in oauth_signature.split(',')])
-
- oauth_signature[u'oauth_nonce'] = oauth_signature.pop(u'OAuth oauth_nonce')
-
- # oauthlib encodes signature with
- # 'Content-Type': 'application/x-www-form-urlencoded'
- # so '='' becomes '%3D'.
- # We send form via browser, so browser will encode it again,
- # So we need to decode signature back:
- oauth_signature[u'oauth_signature'] = six.moves.urllib.parse.unquote(
- oauth_signature[u'oauth_signature']
- )
-
- # Add LTI parameters to OAuth parameters for sending in form.
- lti_parameters.update(oauth_signature)
- return lti_parameters
-
- def get_result(self, user): # pylint: disable=unused-argument
- """
- Helper request handler for GET requests to LTI 2.0 result endpoint
-
- GET handler for lti_2_0_result. Assumes all authorization has been checked.
-
- Arguments:
- request (xblock.django.request.DjangoWebobRequest): Request object (unused)
- real_user (django.contrib.auth.models.User): Actual user linked to anon_id in request path suffix
-
- Returns:
- webob.response: response to this request, in JSON format with status 200 if success
- """
- self.xblock.runtime.rebind_noauth_module_to_user(self.xblock, user)
-
- response = {
- "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
- "@type": "Result"
- }
- if self.xblock.module_score is not None:
- response['resultScore'] = round(self.xblock.module_score, 2)
- response['comment'] = self.xblock.score_comment
-
- return response
-
- def delete_result(self, user): # pylint: disable=unused-argument
- """
- Helper request handler for DELETE requests to LTI 2.0 result endpoint
-
- DELETE handler for lti_2_0_result. Assumes all authorization has been checked.
-
- Arguments:
- request (xblock.django.request.DjangoWebobRequest): Request object (unused)
- real_user (django.contrib.auth.models.User): Actual user linked to anon_id in request path suffix
-
- Returns:
- webob.response: response to this request. status 200 if success
- """
- self.xblock.clear_user_module_score(user)
- return {}
-
- def put_result(self, user, result_json):
- """
- Helper request handler for PUT requests to LTI 2.0 result endpoint
-
- PUT handler for lti_2_0_result. Assumes all authorization has been checked.
-
- Arguments:
- request (xblock.django.request.DjangoWebobRequest): Request object
- real_user (django.contrib.auth.models.User): Actual user linked to anon_id in request path suffix
-
- Returns:
- webob.response: response to this request. status 200 if success. 404 if body of PUT request is malformed
- """
- score, comment = parse_result_json(result_json)
-
- if score is None:
- # According to http://www.imsglobal.org/lti/ltiv2p0/ltiIMGv2p0.html#_Toc361225514
- # PUTting a JSON object with no "resultScore" field is equivalent to a DELETE.
- self.xblock.clear_user_module_score(user)
- else:
- self.xblock.set_user_module_score(user, score, self.xblock.max_score(), comment)
-
- return {}
-
- def verify_result_headers(self, request, verify_content_type=True):
- """
- Helper method to validate LTI 2.0 REST result service HTTP headers. returns if correct, else raises LtiError
-
- Arguments:
- request (xblock.django.request.DjangoWebobRequest): Request object
- verify_content_type (bool): If true, verifies the content type of the request is that spec'ed by LTI 2.0
-
- Returns:
- nothing, but will only return if verification succeeds
-
- Raises:
- LtiError if verification fails
- """
- content_type = request.headers.get('Content-Type')
- if verify_content_type and content_type != LtiConsumer.CONTENT_TYPE_RESULT_JSON:
- log.error("[LTI]: v2.0 result service -- bad Content-Type: %s", content_type)
- error_msg = "For LTI 2.0 result service, Content-Type must be {}. Got {}".format(
- LtiConsumer.CONTENT_TYPE_RESULT_JSON,
- content_type
- )
- raise LtiError(error_msg)
-
- __, secret = self.xblock.lti_provider_key_secret
- try:
- return verify_oauth_body_signature(request, secret, self.xblock.outcome_service_url)
- except (ValueError, LtiError) as err:
- log.error("[LTI]: v2.0 result service -- OAuth body verification failed: %s", str(err))
- raise LtiError(str(err))
diff --git a/lti_consumer/lti_1p1/README.rst b/lti_consumer/lti_1p1/README.rst
new file mode 100644
index 0000000..92410b4
--- /dev/null
+++ b/lti_consumer/lti_1p1/README.rst
@@ -0,0 +1,71 @@
+LTI 1.1 Consumer Class
+-
+
+This is a work in progress implementation of a LTI 1.1 compliant consumer class
+which is request agnostic and can be reused in different contexts (XBlock,
+Django plugin, and even on other frameworks).
+
+This doesn't implement any data storage, just the methods required for handling
+LTI messages.
+
+Also provided is a helper method that can be used to generate an HTML fragment
+which will automatically submit an LTI Launch request once rendered.
+
+Features:
+- LTI 1.1 Launch
+- Support for custom parameters
+
+This implementation was based on the following IMS Global Documents:
+- LTI 1.1 Core Specification: https://www.imsglobal.org/specs/ltiv1p1/
+
+### Using the `lti_embed` helper method
+
+Below is a code snippet of a Django view that will render the HTML fragment
+returned by the `lti_embed` helper method, automatically submit the launch
+request, and redirect to the saLTIre tool.
+
+```python
+from django.contrib.auth.decorators import login_required
+from django.http import HttpResponse
+
+from lti_consumer.lti_1p1.contrib.django import lti_embed
+
+
+@login_required
+def basic_lti_embed(request):
+ """
+ Provides the LTI Embed outside of an xblock
+ """
+
+ return HttpResponse(lti_embed(
+ html_element_id='direct-embed',
+ lti_launch_url='http://lti.tools/saltire/tp',
+ oauth_key='jisc.ac.uk',
+ oauth_secret='secret',
+ resource_link_id='unique-resource-link-id',
+ user_id='student',
+ roles='Student',
+ context_id='some-page',
+ context_title='Some page title',
+ context_label='Some page label',
+ result_sourcedid='unique-result-sourcedid',
+ person_sourcedid=None,
+ person_contact_email_primary=None,
+ outcome_service_url=None,
+ launch_presentation_locale=None
+ ))
+```
+
+To render the LTI Launch within the same webpage as it is launched without
+redirecting the user, simply enclose the template returned by `lti_embed` within
+an `iframe`.
+
+#### Important Note About `lti_embed`
+
+This method uses keyword only arguments as described in
+[PEP-3102](https://www.python.org/dev/peps/pep-3102/). As such, all arguments
+passed to `lti_embed` must use specify the keyword associated to the value.
+Given the large number of arguments for this method, there is a desire to
+guarantee that developers using this method know which arguments are being set
+to which values. This syntax is NOT backwards compatible with python 2.X, but is
+compatible with python 3.5.X or higher.
diff --git a/lti_consumer/lti_1p1/__init__.py b/lti_consumer/lti_1p1/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/lti_consumer/lti_1p1/consumer.py b/lti_consumer/lti_1p1/consumer.py
new file mode 100644
index 0000000..4903439
--- /dev/null
+++ b/lti_consumer/lti_1p1/consumer.py
@@ -0,0 +1,390 @@
+"""
+This module encapsulates code which implements the LTI specification.
+
+For more details see:
+https://www.imsglobal.org/activity/learning-tools-interoperability
+"""
+
+import json
+import logging
+import urllib.parse
+
+from .exceptions import Lti1p1Error
+from .oauth import get_oauth_request_signature, verify_oauth_body_signature
+
+log = logging.getLogger(__name__)
+
+LTI_PARAMETERS = [
+ 'lti_message_type',
+ 'lti_version',
+ 'resource_link_title',
+ 'resource_link_description',
+ 'user_image',
+ 'lis_person_name_given',
+ 'lis_person_name_family',
+ 'lis_person_name_full',
+ 'lis_person_contact_email_primary',
+ 'lis_person_sourcedid',
+ 'role_scope_mentor',
+ 'context_type',
+ 'context_title',
+ 'context_label',
+ 'launch_presentation_locale',
+ 'launch_presentation_document_target',
+ 'launch_presentation_css_url',
+ 'launch_presentation_width',
+ 'launch_presentation_height',
+ 'launch_presentation_return_url',
+ 'tool_consumer_info_product_family_code',
+ 'tool_consumer_info_version',
+ 'tool_consumer_instance_guid',
+ 'tool_consumer_instance_name',
+ 'tool_consumer_instance_description',
+ 'tool_consumer_instance_url',
+ 'tool_consumer_instance_contact_email',
+]
+
+
+def parse_result_json(json_str):
+ """
+ Helper method for verifying LTI 2.0 JSON object contained in the body of the request.
+
+ The json_str must be loadable. It can either be an dict (object) or an array whose first element is an dict,
+ in which case that first dict is considered.
+ The dict must have the "@type" key with value equal to "Result",
+ "resultScore" key with value equal to a number [0, 1], if "resultScore" is not
+ included in the JSON body, score will be returned as None
+ The "@context" key must be present, but we don't do anything with it. And the "comment" key may be
+ present, in which case it must be a string.
+
+ Arguments:
+ json_str (unicode): The body of the LTI 2.0 results service request, which is a JSON string
+
+ Returns:
+ (float, str): (score, [optional]comment) if parsing is successful
+
+ Raises:
+ Lti1p1Error: if verification fails
+ """
+ try:
+ json_obj = json.loads(json_str)
+ except (ValueError, TypeError):
+ msg = "Supplied JSON string in request body could not be decoded: {}".format(json_str)
+ log.error("[LTI] %s", msg)
+ raise Lti1p1Error(msg)
+
+ # The JSON object must be a dict. If a non-empty list is passed in,
+ # use the first element, but only if it is a dict
+ if isinstance(json_obj, list) and len(json_obj) >= 1:
+ json_obj = json_obj[0]
+
+ if not isinstance(json_obj, dict):
+ msg = ("Supplied JSON string is a list that does not contain an object as the first element. {}"
+ .format(json_str))
+ log.error("[LTI] %s", msg)
+ raise Lti1p1Error(msg)
+
+ # '@type' must be "Result"
+ result_type = json_obj.get("@type")
+ if result_type != "Result":
+ msg = "JSON object does not contain correct @type attribute (should be 'Result', is z{})".format(result_type)
+ log.error("[LTI] %s", msg)
+ raise Lti1p1Error(msg)
+
+ # '@context' must be present as a key
+ if '@context' not in json_obj:
+ msg = "JSON object does not contain required key @context"
+ log.error("[LTI] %s", msg)
+ raise Lti1p1Error(msg)
+
+ # Return None if the resultScore key is missing, this condition
+ # will be handled by the upstream caller of this function
+ if "resultScore" not in json_obj:
+ score = None
+ else:
+ # if present, 'resultScore' must be a number between 0 and 1 inclusive
+ try:
+ score = float(json_obj.get('resultScore', "unconvertable")) # Check if float is present and the right type
+ if not 0.0 <= score <= 1.0:
+ msg = 'score value outside the permitted range of 0.0-1.0.'
+ log.error("[LTI] %s", msg)
+ raise Lti1p1Error(msg)
+ except (TypeError, ValueError) as err:
+ msg = "Could not convert resultScore to float: {}".format(str(err))
+ log.error("[LTI] %s", msg)
+ raise Lti1p1Error(msg)
+
+ return score, json_obj.get('comment', "")
+
+
+class LtiConsumer1p1(object): # pylint: disable=bad-option-value, useless-object-inheritance
+ """
+ Limited implementation of the LTI 1.1.
+
+ For the LTI 1.1 specification see:
+ https://www.imsglobal.org/specs/ltiv1p1
+ """
+ CONTENT_TYPE_RESULT_JSON = 'application/vnd.ims.lis.v2.result+json'
+
+ def __init__(self, lti_launch_url, oauth_key, oauth_secret):
+ """
+ Initialize LTI 1.1 Consumer class
+
+ Arguments:
+ lti_launch_url (string): URL to which the LTI Launch should be sent
+ oauth_key (string): OAuth consumer key
+ oauth_secret (string): OAuth consumer secret
+ """
+ self.lti_launch_url = lti_launch_url
+ self.oauth_key = oauth_key
+ self.oauth_secret = oauth_secret
+
+ # IMS LTI data
+ self.lti_user_data = None
+ self.lti_context_data = None
+ self.lti_outcome_service_url = None
+ self.lti_launch_presentation_locale = None
+ self.lti_custom_parameters = None
+
+ def set_user_data(
+ self,
+ user_id,
+ roles,
+ result_sourcedid,
+ person_sourcedid=None,
+ person_contact_email_primary=None
+ ):
+ """
+ Set user data/roles
+
+ Arguments:
+ user_id (string): Unique value identifying the user
+ roles (string): A comma separated list of role values
+ result_sourcedid (string): Indicates the LIS Result Identifier (if any)
+ and uniquely identifies a row and column within the Tool Consumer gradebook
+ person_sourcedid (string): LIS identifier for the user account performing the launch
+ person_contact_email_primary (string): Primary contact email address of the user
+ """
+ self.lti_user_data = {
+ 'user_id': user_id,
+ 'roles': roles,
+ 'lis_result_sourcedid': result_sourcedid,
+ }
+
+ # Additonal user identity data
+ # Optional user data that can be sent to the tool, if the block is configured to do so
+ if person_sourcedid:
+ self.lti_user_data.update({
+ 'lis_person_sourcedid': person_sourcedid,
+ })
+
+ if person_contact_email_primary:
+ self.lti_user_data.update({
+ 'lis_person_contact_email_primary': person_contact_email_primary,
+ })
+
+ def set_context_data(self, context_id, context_title, context_label):
+ """
+ Set LTI context data
+
+ Arguments:
+ context_id (string): Opaque identifier used to uniquely identify the
+ context that contains the link being launched
+ context_title (string): Plain text title of the context
+ context_label (string): Plain text label for the context
+ """
+ self.lti_context_data = {
+ 'context_id': context_id,
+ 'context_title': context_title,
+ 'context_label': context_label,
+ }
+
+ def set_outcome_service_url(self, outcome_service_url):
+ """
+ Set outcome_service_url for scoring
+
+ Arguments:
+ outcome_service_url (string): URL pointing to the outcome service. This
+ is required if the Tool Consumer is accepting outcomes for launches
+ associated with the resource_link_id
+ """
+ self.lti_outcome_service_url = {
+ 'lis_outcome_service_url': outcome_service_url,
+ }
+
+ def set_launch_presentation_locale(self, launch_presentation_locale):
+ """
+ Set launch presentation locale
+
+ Arguments:
+ launch_presentation_locale (string): Language, country and variant as
+ represented using the IETF Best Practices for Tags for Identifying
+ Languages (BCP-47)
+ """
+ self.lti_launch_presentation_locale = {
+ 'launch_presentation_locale': launch_presentation_locale
+ }
+
+ def set_custom_parameters(self, custom_parameters):
+ """
+ Sets custom parameters configured for LTI launch
+
+ Arguments:
+ custom_parameters (dict): Dictionary of custom key/value parameters
+ to be included in the LTI Launch
+
+ Raises:
+ ValueError if custom_parameters is not a dict
+ """
+ if not isinstance(custom_parameters, dict):
+ raise ValueError("Custom parameters must be a key/value dictionary.")
+
+ self.lti_custom_parameters = custom_parameters
+
+ def generate_launch_request(self, resource_link_id):
+ """
+ Signs LTI launch request and returns signature and OAuth parameters.
+
+ Arguments:
+ resource_link_id (string): Opaque identifier guaranteed to be unique
+ for every placement of the link
+
+ Returns:
+ dict: LTI launch parameters
+ """
+
+ # Must have parameters for correct signing from LTI:
+ lti_parameters = {
+ 'oauth_callback': 'about:blank',
+ 'launch_presentation_return_url': '',
+ 'lti_message_type': 'basic-lti-launch-request',
+ 'lti_version': 'LTI-1p0',
+
+ # Parameters required for grading:
+ 'resource_link_id': resource_link_id,
+ }
+
+ # Check if user data is set, then append it to lti message
+ # Raise if isn't set, since some user data is required for the launch
+ if self.lti_user_data:
+ lti_parameters.update(self.lti_user_data)
+ else:
+ raise ValueError("Required user data isn't set.")
+
+ # Check if context data is set, then append it to lti message
+ # Raise if isn't set, since all context data is required for the launch
+ if self.lti_context_data:
+ lti_parameters.update(self.lti_context_data)
+ else:
+ raise ValueError("Required context data isn't set.")
+
+ if self.lti_outcome_service_url:
+ lti_parameters.update(self.lti_outcome_service_url)
+
+ if self.lti_launch_presentation_locale:
+ lti_parameters.update(self.lti_launch_presentation_locale)
+
+ # Appending custom parameter for signing.
+ if self.lti_custom_parameters:
+ lti_parameters.update(self.lti_custom_parameters)
+
+ headers = {
+ # This is needed for body encoding:
+ 'Content-Type': 'application/x-www-form-urlencoded',
+ }
+
+ oauth_signature = get_oauth_request_signature(
+ self.oauth_key,
+ self.oauth_secret,
+ self.lti_launch_url,
+ headers,
+ lti_parameters
+ )
+
+ # Parse headers to pass to template as part of context:
+ oauth_signature = dict([param.strip().replace('"', '').split('=') for param in oauth_signature.split(',')])
+
+ oauth_signature[u'oauth_nonce'] = oauth_signature.pop(u'OAuth oauth_nonce')
+
+ # oauthlib encodes signature with
+ # 'Content-Type': 'application/x-www-form-urlencoded'
+ # so '='' becomes '%3D'.
+ # We send form via browser, so browser will encode it again,
+ # So we need to decode signature back:
+ oauth_signature[u'oauth_signature'] = urllib.parse.unquote(
+ oauth_signature[u'oauth_signature']
+ )
+
+ # Add LTI parameters to OAuth parameters for sending in form.
+ lti_parameters.update(oauth_signature)
+ return lti_parameters
+
+ def get_result(self, result_score=None, score_comment=None): # pylint: disable=unused-argument
+ """
+ Returns response body for GET requests to LTI 2.0 result endpoint
+
+ Arguments:
+ result_score (float): The result score of the user
+ score_comment (string): A text comment describing the score
+
+ Returns:
+ dict: response to this request, in JSON format with resultScore and comment if provided
+ """
+ response = {
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "@type": "Result"
+ }
+ if result_score is not None:
+ response['resultScore'] = round(result_score, 2)
+ response['comment'] = score_comment
+
+ return response
+
+ def delete_result(self):
+ """
+ Returns response body for DELETE requests to LTI 2.0 result endpoint
+ """
+ return {}
+
+ def put_result(self):
+ """
+ Returns response body for PUT requests to LTI 2.0 result endpoint
+ """
+ return {}
+
+ def verify_result_headers(self, request, verify_content_type=True):
+ """
+ Helper method to validate LTI 2.0 REST result service HTTP headers. returns if correct, else raises Lti1p1Error
+
+ Arguments:
+ request (webob.Request): Request object
+ verify_content_type (bool): If true, verifies the content type of the request is that spec'ed by LTI 2.0
+
+ Returns:
+ nothing, but will only return if verification succeeds
+
+ Raises:
+ Lti1p1Error if verification fails
+ """
+ content_type = request.headers.get('Content-Type')
+ if verify_content_type and content_type != LtiConsumer1p1.CONTENT_TYPE_RESULT_JSON:
+ log.error("[LTI]: v2.0 result service -- bad Content-Type: %s", content_type)
+ error_msg = "For LTI 2.0 result service, Content-Type must be {}. Got {}".format(
+ LtiConsumer1p1.CONTENT_TYPE_RESULT_JSON,
+ content_type
+ )
+ raise Lti1p1Error(error_msg)
+
+ # Check if scoring data is set, then append it to lti message
+ # Raise if isn't set, since some scoring data is required for the launch
+ if self.lti_outcome_service_url:
+ outcome_service_url = self.lti_outcome_service_url['lis_outcome_service_url']
+ else:
+ log.error("[LTI]: v2.0 result service -- lis_outcome_service_url not set")
+ raise ValueError("Required outcome_service_url not set.")
+
+ try:
+ return verify_oauth_body_signature(request, self.oauth_secret, outcome_service_url)
+ except (ValueError, Lti1p1Error) as err:
+ log.error("[LTI]: v2.0 result service -- OAuth body verification failed: %s", str(err))
+ raise Lti1p1Error(str(err))
diff --git a/lti_consumer/lti_1p1/contrib/__init__.py b/lti_consumer/lti_1p1/contrib/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/lti_consumer/lti_1p1/contrib/django.py b/lti_consumer/lti_1p1/contrib/django.py
new file mode 100644
index 0000000..137c5e7
--- /dev/null
+++ b/lti_consumer/lti_1p1/contrib/django.py
@@ -0,0 +1,123 @@
+"""
+This module provides functionality for rendering an LTI embed without an XBlock.
+"""
+
+# See comment in docstring for explanation of the usage of ResourceLoader
+from xblockutils.resources import ResourceLoader
+
+from ..consumer import LtiConsumer1p1
+
+
+def lti_embed(
+ *,
+ html_element_id,
+ lti_launch_url,
+ oauth_key,
+ oauth_secret,
+ resource_link_id,
+ user_id,
+ roles,
+ context_id,
+ context_title,
+ context_label,
+ result_sourcedid,
+ person_sourcedid=None,
+ person_contact_email_primary=None,
+ outcome_service_url=None,
+ launch_presentation_locale=None,
+ **custom_parameters
+):
+ """
+ Returns an HTML template with JavaScript that will launch an LTI embed
+
+ IMPORTANT NOTE: This method uses keyword only arguments as described in PEP 3102.
+ Given the large number of arguments for this method, there is a desire to
+ guarantee that developers using this method know which arguments are being set
+ to which values.
+ See https://www.python.org/dev/peps/pep-3102/
+
+ This method will use the LtiConsumer1p1 class to generate an HTML form and
+ JavaScript that will automatically launch the LTI embedding, but it does not
+ generate any response to encapsulate this content. The caller of this method
+ must render the HTML on their own.
+
+ Note: This method uses xblockutils.resources.ResourceLoader to load the HTML
+ template used. The rationale for this is that ResourceLoader is agnostic
+ to XBlock code and functionality. It is recommended that this remain in use
+ until LTI1.3 support is merged, or a better means of loading the template is
+ made available.
+
+ Arguments:
+ html_element_id (string): Value to use as the HTML element id in the HTML form
+ lti_launch_url (string): The URL to send the LTI Launch request to
+ oauth_key (string): The OAuth consumer key
+ oauth_secret (string): The OAuth consumer secret
+ resource_link_id (string): Opaque identifier guaranteed to be unique
+ for every placement of the link
+ user_id (string): Unique value identifying the user
+ roles (string): A comma separated list of role values
+ context_id (string): Opaque identifier used to uniquely identify the
+ context that contains the link being launched
+ context_title (string): Plain text title of the context
+ context_label (string): Plain text label for the context
+ result_sourcedid (string): Indicates the LIS Result Identifier (if any)
+ and uniquely identifies a row and column within the Tool Consumer gradebook
+ person_sourcedid (string): LIS identifier for the user account performing the launch
+ person_contact_email_primary (string): Primary contact email address of the user
+ outcome_service_url (string): URL pointing to the outcome service. This
+ is required if the Tool Consumer is accepting outcomes for launches
+ associated with the resource_link_id
+ launch_presentation_locale (string): Language, country and variant as
+ represented using the IETF Best Practices for Tags for Identifying
+ Languages (BCP-47)
+ custom_parameters (dict): Contains any other keyword arguments not listed
+ above. It will filter out all arguments provided that do not start with
+ 'custom_' and will submit the remaining arguments on the LTI Launch form
+
+ Returns:
+ unicode: HTML template with the form and JavaScript to automatically
+ launch the LTI embedding
+ """
+ lti_consumer = LtiConsumer1p1(lti_launch_url, oauth_key, oauth_secret)
+
+ # Set LTI parameters from kwargs
+ lti_consumer.set_user_data(
+ user_id,
+ roles,
+ result_sourcedid,
+ person_sourcedid=person_sourcedid,
+ person_contact_email_primary=person_contact_email_primary
+ )
+ lti_consumer.set_context_data(
+ context_id,
+ context_title,
+ context_label
+ )
+
+ if outcome_service_url:
+ lti_consumer.set_outcome_service_url(outcome_service_url)
+
+ if launch_presentation_locale:
+ lti_consumer.set_launch_presentation_locale(launch_presentation_locale)
+
+ lti_consumer.set_custom_parameters(
+ {
+ key: value
+ for key, value in custom_parameters.items()
+ if key.startswith('custom_')
+ }
+ )
+
+ lti_parameters = lti_consumer.generate_launch_request(resource_link_id)
+
+ # Prepare form data
+ context = {
+ 'launch_url': lti_launch_url,
+ 'element_id': html_element_id
+ }
+ context.update({'lti_parameters': lti_parameters})
+
+ # Render the form template and return the template
+ loader = ResourceLoader(__name__)
+ template = loader.render_mako_template('../../templates/html/lti_launch.html', context)
+ return template
diff --git a/lti_consumer/lti_1p1/contrib/tests/__init__.py b/lti_consumer/lti_1p1/contrib/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/lti_consumer/lti_1p1/contrib/tests/test_django.py b/lti_consumer/lti_1p1/contrib/tests/test_django.py
new file mode 100644
index 0000000..c958584
--- /dev/null
+++ b/lti_consumer/lti_1p1/contrib/tests/test_django.py
@@ -0,0 +1,136 @@
+# -*- coding: utf-8 -*-
+"""
+Unit tests for lti_consumer.lti module
+"""
+
+from django.test.testcases import TestCase
+from mock import Mock, patch, ANY
+
+from lti_consumer.lti_1p1.contrib.django import lti_embed
+
+
+class TestLtiEmbed(TestCase):
+ """
+ Unit tests for contrib.django.lti_embed
+ """
+
+ def setUp(self):
+ super(TestLtiEmbed, self).setUp()
+ self.html_element_id = 'html_element_id'
+ self.lti_launch_url = 'lti_launch_url'
+ self.oauth_key = 'oauth_key'
+ self.oauth_secret = 'oauth_secret'
+ self.resource_link_id = 'resource_link_id'
+ self.user_id = 'user_id'
+ self.roles = 'roles'
+ self.context_id = 'context_id'
+ self.context_title = 'context_title'
+ self.context_label = 'context_label'
+ self.result_sourcedid = 'result_sourcedid'
+
+ def test_non_keyword_arguments_raise_type_error(self):
+ with self.assertRaises(TypeError):
+ lti_embed( # pylint: disable=too-many-function-args,missing-kwoa
+ self.html_element_id,
+ self.lti_launch_url,
+ self.oauth_key,
+ self.oauth_secret,
+ self.resource_link_id,
+ self.user_id,
+ self.roles,
+ self.context_id,
+ self.context_title,
+ self.context_label,
+ self.result_sourcedid
+ )
+
+ def test_missing_required_arguments_raise_type_error(self):
+ with self.assertRaises(TypeError):
+ # Missing result_sourcedid
+ lti_embed( # pylint: disable=missing-kwoa
+ html_element_id=self.html_element_id,
+ lti_launch_url=self.lti_launch_url,
+ oauth_key=self.oauth_key,
+ oauth_secret=self.oauth_secret,
+ resource_link_id=self.resource_link_id,
+ user_id=self.user_id,
+ roles=self.roles,
+ context_id=self.context_id,
+ context_title=self.context_title,
+ context_label=self.context_label
+ )
+
+ @patch('lti_consumer.lti_1p1.contrib.django.LtiConsumer1p1')
+ def test_consumer_initialized_properly(self, mock_lti_consumer_class):
+ lti_embed(
+ html_element_id=self.html_element_id,
+ lti_launch_url=self.lti_launch_url,
+ oauth_key=self.oauth_key,
+ oauth_secret=self.oauth_secret,
+ resource_link_id=self.resource_link_id,
+ user_id=self.user_id,
+ roles=self.roles,
+ context_id=self.context_id,
+ context_title=self.context_title,
+ context_label=self.context_label,
+ result_sourcedid=self.result_sourcedid
+ )
+
+ mock_lti_consumer_class.assert_called_with(self.lti_launch_url, self.oauth_key, self.oauth_secret)
+
+ @patch('lti_consumer.lti_1p1.contrib.django.LtiConsumer1p1.set_custom_parameters')
+ @patch('lti_consumer.lti_1p1.contrib.django.LtiConsumer1p1.generate_launch_request', Mock(return_value={}))
+ def test_custom_parameters_ignore_keyword_args_without_custom_prefix(self, mock_set_custom_parameters):
+ lti_embed(
+ html_element_id=self.html_element_id,
+ lti_launch_url=self.lti_launch_url,
+ oauth_key=self.oauth_key,
+ oauth_secret=self.oauth_secret,
+ resource_link_id=self.resource_link_id,
+ user_id=self.user_id,
+ roles=self.roles,
+ context_id=self.context_id,
+ context_title=self.context_title,
+ context_label=self.context_label,
+ result_sourcedid=self.result_sourcedid,
+ custom_parameter_1='custom_parameter_1',
+ custom_parameter_2='custom_parameter_2',
+ parameter_3='parameter_3',
+ )
+
+ expected_custom_parameters = {
+ 'custom_parameter_1': 'custom_parameter_1',
+ 'custom_parameter_2': 'custom_parameter_2'
+ }
+ mock_set_custom_parameters.assert_called_with(expected_custom_parameters)
+
+ @patch('lti_consumer.lti_1p1.contrib.django.LtiConsumer1p1.generate_launch_request', Mock(return_value={'a': 1}))
+ @patch('lti_consumer.lti_1p1.contrib.django.ResourceLoader.render_mako_template')
+ def test_make_template_rendered_with_correct_context_and_returned(self, mock_render_mako_template):
+ fake_template = 'SOME_TEMPLATE'
+ mock_render_mako_template.return_value = fake_template
+
+ rendered_template = lti_embed(
+ html_element_id=self.html_element_id,
+ lti_launch_url=self.lti_launch_url,
+ oauth_key=self.oauth_key,
+ oauth_secret=self.oauth_secret,
+ resource_link_id=self.resource_link_id,
+ user_id=self.user_id,
+ roles=self.roles,
+ context_id=self.context_id,
+ context_title=self.context_title,
+ context_label=self.context_label,
+ result_sourcedid=self.result_sourcedid,
+ custom_parameter_1='custom_parameter_1',
+ custom_parameter_2='custom_parameter_2',
+ parameter_3='parameter_3',
+ )
+
+ expected_context = {
+ 'element_id': self.html_element_id,
+ 'launch_url': self.lti_launch_url,
+ 'lti_parameters': {'a': 1}
+ }
+ mock_render_mako_template.assert_called_with(ANY, expected_context)
+ self.assertEqual(rendered_template, fake_template)
diff --git a/lti_consumer/lti_1p1/exceptions.py b/lti_consumer/lti_1p1/exceptions.py
new file mode 100644
index 0000000..80383ae
--- /dev/null
+++ b/lti_consumer/lti_1p1/exceptions.py
@@ -0,0 +1,9 @@
+"""
+Exceptions for the LTI1.1 Consumer.
+"""
+
+
+class Lti1p1Error(Exception):
+ """
+ General error class for LTI1.1 Consumer usage.
+ """
diff --git a/lti_consumer/oauth.py b/lti_consumer/lti_1p1/oauth.py
similarity index 79%
rename from lti_consumer/oauth.py
rename to lti_consumer/lti_1p1/oauth.py
index 8bd5d89..61f8daa 100644
--- a/lti_consumer/oauth.py
+++ b/lti_consumer/lti_1p1/oauth.py
@@ -5,13 +5,11 @@ Utility functions for working with OAuth signatures.
import base64
import hashlib
import logging
+import urllib.parse
-import six.moves.urllib.error
-import six.moves.urllib.parse
-import six
from oauthlib import oauth1
-from .exceptions import LtiError
+from .exceptions import Lti1p1Error
log = logging.getLogger(__name__)
@@ -46,20 +44,20 @@ def get_oauth_request_signature(key, secret, url, headers, body):
Returns:
str: Authorization header for the OAuth signed request
"""
- client = oauth1.Client(client_key=six.text_type(key), client_secret=six.text_type(secret))
+ client = oauth1.Client(client_key=str(key), client_secret=str(secret))
try:
# Add Authorization header which looks like:
# Authorization: OAuth oauth_nonce="80966668944732164491378916897",
# oauth_timestamp="1378916897", oauth_version="1.0", oauth_signature_method="HMAC-SHA1",
# oauth_consumer_key="", oauth_signature="frVp4JuvT1mVXlxktiAUjQ7%2F1cw%3D"
_, headers, _ = client.sign(
- six.text_type(url.strip()),
+ str(url.strip()),
http_method=u'POST',
body=body,
headers=headers
)
except ValueError: # Scheme not in url.
- raise LtiError("Failed to sign oauth request")
+ raise Lti1p1Error("Failed to sign oauth request")
return headers['Authorization']
@@ -74,17 +72,17 @@ def verify_oauth_body_signature(request, lti_provider_secret, service_url):
with content types other than application/x-www-form-urlencoded.
Arguments:
- request (xblock.django.request.DjangoWebobRequest): Request object for current HTTP request
+ request (webob.Request): Request object for current HTTP request
lti_provider_secret (str): Secret key for the LTI provider
service_url (str): URL that the request was made to
content_type (str): HTTP content type of the request
Raises:
- LtiError if request is incorrect.
+ Lti1p1Error if request is incorrect.
"""
headers = {
- 'Authorization': six.text_type(request.headers.get('Authorization')),
+ 'Authorization': str(request.headers.get('Authorization')),
'Content-Type': request.content_type,
}
@@ -95,14 +93,14 @@ def verify_oauth_body_signature(request, lti_provider_secret, service_url):
oauth_headers = dict(oauth_params)
oauth_signature = oauth_headers.pop('oauth_signature')
mock_request_lti_1 = SignedRequest(
- uri=six.text_type(six.moves.urllib.parse.unquote(service_url)),
- http_method=six.text_type(request.method),
+ uri=str(urllib.parse.unquote(service_url)),
+ http_method=str(request.method),
params=list(oauth_headers.items()),
signature=oauth_signature
)
mock_request_lti_2 = SignedRequest(
- uri=six.text_type(six.moves.urllib.parse.unquote(request.url)),
- http_method=six.text_type(request.method),
+ uri=str(urllib.parse.unquote(request.url)),
+ http_method=str(request.method),
params=list(oauth_headers.items()),
signature=oauth_signature
)
@@ -115,7 +113,7 @@ def verify_oauth_body_signature(request, lti_provider_secret, service_url):
service_url,
request.body
)
- raise LtiError("OAuth body hash verification is failed.")
+ raise Lti1p1Error("OAuth body hash verification has failed.")
if (not oauth1.rfc5849.signature.verify_hmac_sha1(mock_request_lti_1, lti_provider_secret) and not
oauth1.rfc5849.signature.verify_hmac_sha1(mock_request_lti_2, lti_provider_secret)):
@@ -124,9 +122,9 @@ def verify_oauth_body_signature(request, lti_provider_secret, service_url):
"headers:%s url:%s method:%s",
oauth_headers,
service_url,
- six.text_type(request.method)
+ str(request.method)
)
- raise LtiError("OAuth signature verification has failed.")
+ raise Lti1p1Error("OAuth signature verification has failed.")
return True
@@ -139,25 +137,25 @@ def log_authorization_header(request, client_key, client_secret):
the request header and body according to OAuth 1 Body signing
Arguments:
- request (xblock.django.request.DjangoWebobRequest): Request object to log Authorization header for
+ request (webob.Request): Request object to log Authorization header for
Returns:
nothing
"""
sha1 = hashlib.sha1()
sha1.update(request.body)
- oauth_body_hash = six.text_type(base64.b64encode(sha1.digest())) # pylint: disable=too-many-function-args
+ oauth_body_hash = str(base64.b64encode(sha1.digest())) # pylint: disable=too-many-function-args
log.debug("[LTI] oauth_body_hash = %s", oauth_body_hash)
client = oauth1.Client(client_key, client_secret)
params = client.get_oauth_params(request)
params.append((u'oauth_body_hash', oauth_body_hash))
mock_request = SignedRequest(
- uri=six.text_type(six.moves.urllib.parse.unquote(request.url)),
+ uri=str(urllib.parse.unquote(request.url)),
headers=request.headers,
body=u"",
decoded_body=u"",
oauth_params=params,
- http_method=six.text_type(request.method),
+ http_method=str(request.method),
)
sig = client.get_oauth_signature(mock_request)
mock_request.oauth_params.append((u'oauth_signature', sig))
diff --git a/lti_consumer/lti_1p1/tests/__init__.py b/lti_consumer/lti_1p1/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/lti_consumer/lti_1p1/tests/test_consumer.py b/lti_consumer/lti_1p1/tests/test_consumer.py
new file mode 100644
index 0000000..06389b0
--- /dev/null
+++ b/lti_consumer/lti_1p1/tests/test_consumer.py
@@ -0,0 +1,349 @@
+# -*- coding: utf-8 -*-
+"""
+Unit tests for lti_consumer.lti_1p1.consumer module
+"""
+
+import unittest
+
+from mock import Mock, patch
+
+from lti_consumer.lti_1p1.exceptions import Lti1p1Error
+from lti_consumer.lti_1p1.consumer import LtiConsumer1p1, parse_result_json
+from lti_consumer.tests.unit.test_utils import make_request
+
+INVALID_JSON_INPUTS = [
+ ([
+ u"kk", # ValueError
+ u"{{}", # ValueError
+ u"{}}", # ValueError
+ 3, # TypeError
+ {}, # TypeError
+ ], u"Supplied JSON string in request body could not be decoded"),
+ ([
+ u"3", # valid json, not array or object
+ u"[]", # valid json, array too small
+ u"[3, {}]", # valid json, 1st element not an object
+ ], u"Supplied JSON string is a list that does not contain an object as the first element"),
+ ([
+ u'{"@type": "NOTResult"}', # @type key must have value 'Result'
+ ], u"JSON object does not contain correct @type attribute"),
+ ([
+ # @context missing
+ u'{"@type": "Result", "resultScore": 0.1}',
+ ], u"JSON object does not contain required key"),
+ ([
+ u'''
+ {"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "resultScore": 100}''' # score out of range
+ ], u"score value outside the permitted range of 0.0-1.0."),
+ ([
+ u'''
+ {"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "resultScore": -2}''' # score out of range
+ ], u"score value outside the permitted range of 0.0-1.0."),
+ ([
+ u'''
+ {"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "resultScore": "1b"}''', # score ValueError
+ u'''
+ {"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "resultScore": {}}''', # score TypeError
+ ], u"Could not convert resultScore to float"),
+]
+
+VALID_JSON_INPUTS = [
+ (u'''
+ {"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "resultScore": 0.1}''', 0.1, u""), # no comment means we expect ""
+ (u'''
+ [{"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "@id": "anon_id:abcdef0123456789",
+ "resultScore": 0.1}]''', 0.1, u""), # OK to have array of objects -- just take the first. @id is okay too
+ (u'''
+ {"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "resultScore": 0.1,
+ "comment": "ಠ益ಠ"}''', 0.1, u"ಠ益ಠ"), # unicode comment
+ (u'''
+ {"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result"}''', None, u""), # no score means we expect None
+ (u'''
+ {"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "resultScore": 0.0}''', 0.0, u""), # test lower score boundary
+ (u'''
+ {"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "resultScore": 1.0}''', 1.0, u""), # test upper score boundary
+]
+
+GET_RESULT_RESPONSE = {
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "@type": "Result",
+}
+
+
+class TestParseResultJson(unittest.TestCase):
+ """
+ Unit tests for `lti_consumer.lti_1p1.consumer.parse_result_json`
+ """
+
+ def test_invalid_json(self):
+ """
+ Test invalid json raises exception
+ """
+ for error_inputs, error_message in INVALID_JSON_INPUTS:
+ for error_input in error_inputs:
+ with self.assertRaisesRegex(Lti1p1Error, error_message):
+ parse_result_json(error_input)
+
+ def test_valid_json(self):
+ """
+ Test valid json returns expected values
+ """
+ for json_str, expected_score, expected_comment in VALID_JSON_INPUTS:
+ score, comment = parse_result_json(json_str)
+ self.assertEqual(score, expected_score)
+ self.assertEqual(comment, expected_comment)
+
+
+class TestLtiConsumer1p1(unittest.TestCase):
+ """
+ Unit tests for LtiConsumer
+ """
+
+ def setUp(self):
+ super(TestLtiConsumer1p1, self).setUp()
+ self.lti_launch_url = 'lti_launch_url'
+ self.oauth_key = 'fake_consumer_key'
+ self.oauth_secret = 'fake_signature'
+ self.lti_consumer = LtiConsumer1p1(self.lti_launch_url, self.oauth_key, self.oauth_secret)
+
+ def test_set_custom_parameters_with_non_dict_raises_error(self):
+ with self.assertRaises(ValueError):
+ self.lti_consumer.set_custom_parameters('custom_value')
+
+ def test_generate_launch_request_with_no_user_data_raises_error(self):
+ with self.assertRaises(ValueError):
+ self.lti_consumer.generate_launch_request('resource_link_id')
+
+ def test_generate_launch_request_with_no_context_data_raises_error(self):
+ self.lti_consumer.set_user_data('user_id', 'roles', 'result_sourcedid')
+ with self.assertRaises(ValueError):
+ self.lti_consumer.generate_launch_request('resource_link_id')
+
+ @patch(
+ 'lti_consumer.lti_1p1.consumer.get_oauth_request_signature',
+ Mock(return_value=(
+ 'OAuth oauth_nonce="fake_nonce", '
+ 'oauth_timestamp="fake_timestamp", oauth_version="fake_version", oauth_signature_method="fake_method", '
+ 'oauth_consumer_key="fake_consumer_key", oauth_signature="fake_signature"'
+ ))
+ )
+ def test_generate_launch_request_with_user_and_context_data_succeeds(self):
+ user_id = 'user_id'
+ roles = 'roles'
+ result_sourcedid = 'result_sourcedid'
+ context_id = 'context_id'
+ context_title = 'context_title'
+ context_label = 'context_label'
+ resource_link_id = 'resource_link_id'
+
+ self.lti_consumer.set_user_data(user_id, roles, result_sourcedid)
+ self.lti_consumer.set_context_data(context_id, context_title, context_label)
+
+ lti_parameters = self.lti_consumer.generate_launch_request(resource_link_id)
+
+ expected_lti_parameters = {
+ 'oauth_callback': 'about:blank',
+ 'launch_presentation_return_url': '',
+ 'lti_message_type': 'basic-lti-launch-request',
+ 'lti_version': 'LTI-1p0',
+ 'user_id': user_id,
+ 'roles': roles,
+ 'lis_result_sourcedid': result_sourcedid,
+ 'context_id': context_id,
+ 'context_label': context_label,
+ 'context_title': context_title,
+ 'resource_link_id': resource_link_id,
+ 'oauth_nonce': 'fake_nonce',
+ 'oauth_timestamp': 'fake_timestamp',
+ 'oauth_version': 'fake_version',
+ 'oauth_signature_method': 'fake_method',
+ 'oauth_consumer_key': 'fake_consumer_key',
+ 'oauth_signature': 'fake_signature',
+ }
+ self.assertEqual(lti_parameters, expected_lti_parameters)
+
+ @patch(
+ 'lti_consumer.lti_1p1.consumer.get_oauth_request_signature',
+ Mock(return_value=(
+ 'OAuth oauth_nonce="fake_nonce", '
+ 'oauth_timestamp="fake_timestamp", oauth_version="fake_version", oauth_signature_method="fake_method", '
+ 'oauth_consumer_key="fake_consumer_key", oauth_signature="fake_signature"'
+ ))
+ )
+ def test_generate_launch_request_with_all_optional_parameters_set_succeeds(self):
+ user_id = 'user_id'
+ roles = 'roles'
+ result_sourcedid = 'result_sourcedid'
+ person_sourcedid = 'person_sourcedid'
+ person_contact_email_primary = 'person_contact_email_primary'
+ context_id = 'context_id'
+ context_title = 'context_title'
+ context_label = 'context_label'
+ outcome_service_url = 'outcome_service_url'
+ launch_presentation_locale = 'launch_presentation_locale'
+ custom_parameters = {
+ 'custom_parameter_1': 'custom1',
+ 'custom_parameter_2': 'custom2',
+ }
+ resource_link_id = 'resource_link_id'
+
+ self.lti_consumer.set_user_data(
+ user_id,
+ roles,
+ result_sourcedid,
+ person_sourcedid=person_sourcedid,
+ person_contact_email_primary=person_contact_email_primary
+ )
+ self.lti_consumer.set_context_data(context_id, context_title, context_label)
+ self.lti_consumer.set_outcome_service_url(outcome_service_url)
+ self.lti_consumer.set_launch_presentation_locale(launch_presentation_locale)
+ self.lti_consumer.set_custom_parameters(custom_parameters)
+
+ lti_parameters = self.lti_consumer.generate_launch_request(resource_link_id)
+
+ expected_lti_parameters = {
+ 'oauth_callback': 'about:blank',
+ 'launch_presentation_return_url': '',
+ 'lti_message_type': 'basic-lti-launch-request',
+ 'lti_version': 'LTI-1p0',
+ 'user_id': user_id,
+ 'roles': roles,
+ 'lis_result_sourcedid': result_sourcedid,
+ 'lis_person_sourcedid': person_sourcedid,
+ 'lis_person_contact_email_primary': person_contact_email_primary,
+ 'context_id': context_id,
+ 'context_label': context_label,
+ 'context_title': context_title,
+ 'lis_outcome_service_url': outcome_service_url,
+ 'launch_presentation_locale': launch_presentation_locale,
+ 'custom_parameter_1': custom_parameters['custom_parameter_1'],
+ 'custom_parameter_2': custom_parameters['custom_parameter_2'],
+ 'resource_link_id': resource_link_id,
+ 'oauth_nonce': 'fake_nonce',
+ 'oauth_timestamp': 'fake_timestamp',
+ 'oauth_version': 'fake_version',
+ 'oauth_signature_method': 'fake_method',
+ 'oauth_consumer_key': 'fake_consumer_key',
+ 'oauth_signature': 'fake_signature',
+ }
+ self.assertEqual(lti_parameters, expected_lti_parameters)
+
+ def test_get_result_with_no_score_or_comment(self):
+ self.assertEqual(self.lti_consumer.get_result(), GET_RESULT_RESPONSE)
+
+ def test_get_result_with_score_and_comment(self):
+ score = 1.234
+ comment = 'score_comment'
+
+ full_response = GET_RESULT_RESPONSE
+ full_response.update({
+ 'resultScore': 1.23,
+ 'comment': comment
+ })
+ self.assertEqual(self.lti_consumer.get_result(score, comment), full_response)
+
+ def test_put_result(self):
+ self.assertEqual(self.lti_consumer.put_result(), {})
+
+ def test_delete_result(self):
+ self.assertEqual(self.lti_consumer.delete_result(), {})
+
+ @patch('lti_consumer.lti_1p1.consumer.log')
+ def test_verify_result_headers_verify_content_type_true(self, mock_log):
+ """
+ Test wrong content type raises exception if `verify_content_type` is True
+ """
+ request = make_request('')
+
+ with self.assertRaises(Lti1p1Error):
+ self.lti_consumer.verify_result_headers(request)
+
+ assert mock_log.error.called
+
+ @patch('lti_consumer.lti_1p1.consumer.log')
+ def test_verify_result_headers_no_outcome_service_url(self, mock_log):
+ """
+ Test exception raised if no outcome_service_url is set
+ """
+ request = make_request('')
+ request.environ['CONTENT_TYPE'] = LtiConsumer1p1.CONTENT_TYPE_RESULT_JSON
+
+ with self.assertRaises(ValueError):
+ self.lti_consumer.verify_result_headers(request)
+
+ assert mock_log.error.called
+
+ @patch('lti_consumer.lti_1p1.consumer.verify_oauth_body_signature', Mock(side_effect=Lti1p1Error))
+ @patch('lti_consumer.lti_1p1.consumer.log')
+ def test_verify_result_headers_lti_error(self, mock_log):
+ """
+ Test exception raised if request header verification raises error
+ """
+ request = make_request('')
+ request.environ['CONTENT_TYPE'] = LtiConsumer1p1.CONTENT_TYPE_RESULT_JSON
+
+ self.lti_consumer.set_outcome_service_url('outcome_service_url')
+ with self.assertRaises(Lti1p1Error):
+ self.lti_consumer.verify_result_headers(request)
+
+ assert mock_log.error.called
+
+ @patch('lti_consumer.lti_1p1.consumer.verify_oauth_body_signature', Mock(side_effect=ValueError))
+ @patch('lti_consumer.lti_1p1.consumer.log')
+ def test_verify_result_headers_value_error(self, mock_log):
+ """
+ Test exception raised if request header verification raises error
+ """
+ request = make_request('')
+ request.environ['CONTENT_TYPE'] = LtiConsumer1p1.CONTENT_TYPE_RESULT_JSON
+
+ self.lti_consumer.set_outcome_service_url('outcome_service_url')
+ with self.assertRaises(Lti1p1Error):
+ self.lti_consumer.verify_result_headers(request)
+
+ assert mock_log.error.called
+
+ @patch('lti_consumer.lti_1p1.consumer.verify_oauth_body_signature', Mock(return_value=True))
+ def test_verify_result_headers_valid(self):
+ """
+ Test True is returned if request is valid
+ """
+ request = make_request('')
+ request.environ['CONTENT_TYPE'] = LtiConsumer1p1.CONTENT_TYPE_RESULT_JSON
+
+ self.lti_consumer.set_outcome_service_url('outcome_service_url')
+ response = self.lti_consumer.verify_result_headers(request)
+
+ self.assertTrue(response)
+
+ @patch('lti_consumer.lti_1p1.consumer.verify_oauth_body_signature', Mock(return_value=True))
+ def test_verify_result_headers_verify_content_type_false_valid(self):
+ """
+ Test content type check skipped if `verify_content_type` is False
+ """
+ request = make_request('')
+ request.environ['CONTENT_TYPE'] = LtiConsumer1p1.CONTENT_TYPE_RESULT_JSON
+
+ self.lti_consumer.set_outcome_service_url('outcome_service_url')
+ response = self.lti_consumer.verify_result_headers(request, False)
+
+ self.assertTrue(response)
diff --git a/lti_consumer/tests/unit/test_oauth.py b/lti_consumer/lti_1p1/tests/test_oauth.py
similarity index 88%
rename from lti_consumer/tests/unit/test_oauth.py
rename to lti_consumer/lti_1p1/tests/test_oauth.py
index 738d7a6..d7ef35b 100644
--- a/lti_consumer/tests/unit/test_oauth.py
+++ b/lti_consumer/lti_1p1/tests/test_oauth.py
@@ -6,10 +6,10 @@ import unittest
from mock import Mock, patch
-from lti_consumer.exceptions import LtiError
-from lti_consumer.oauth import (get_oauth_request_signature,
- log_authorization_header,
- verify_oauth_body_signature)
+from lti_consumer.lti_1p1.exceptions import Lti1p1Error
+from lti_consumer.lti_1p1.oauth import (get_oauth_request_signature,
+ log_authorization_header,
+ verify_oauth_body_signature)
from lti_consumer.tests.unit.test_utils import make_request
OAUTH_PARAMS = [
@@ -46,7 +46,7 @@ class TestGetOauthRequestSignature(unittest.TestCase):
"""
mock_client_sign.side_effect = ValueError
- with self.assertRaises(LtiError):
+ with self.assertRaises(Lti1p1Error):
__ = get_oauth_request_signature('test', 'secret', '', {}, '')
@@ -69,7 +69,7 @@ class TestVerifyOauthBodySignature(unittest.TestCase):
"""
Test exception is raised when the request signature is invalid
"""
- with self.assertRaises(LtiError):
+ with self.assertRaises(Lti1p1Error):
verify_oauth_body_signature(make_request(''), 'test', 'secret')
@patch('oauthlib.oauth1.rfc5849.signature.verify_hmac_sha1', Mock(return_value=False))
@@ -78,7 +78,7 @@ class TestVerifyOauthBodySignature(unittest.TestCase):
"""
Test exception is raised when the request signature is missing oauth_body_hash
"""
- with self.assertRaises(LtiError):
+ with self.assertRaises(Lti1p1Error):
verify_oauth_body_signature(make_request(''), 'test', 'secret')
@@ -87,7 +87,7 @@ class TestLogCorrectAuthorizationHeader(unittest.TestCase):
Unit tests for `lti_consumer.oauth.log_authorization_header`
"""
- @patch('lti_consumer.oauth.log')
+ @patch('lti_consumer.lti_1p1.oauth.log')
def test_log_auth_header(self, mock_log):
"""
Test that log.debug is called
diff --git a/lti_consumer/lti_xblock.py b/lti_consumer/lti_xblock.py
index 5384111..f5b923b 100644
--- a/lti_consumer/lti_xblock.py
+++ b/lti_consumer/lti_xblock.py
@@ -50,29 +50,28 @@ What is supported:
GET / PUT / DELETE HTTP methods respectively
"""
-from __future__ import absolute_import, unicode_literals
-
import logging
import re
+import urllib.parse
import uuid
from collections import namedtuple
from importlib import import_module
-import six
-from six.moves.urllib import parse
-from django.utils import timezone
import bleach
+from django.utils import timezone
+from web_fragments.fragment import Fragment
+
from Crypto.PublicKey import RSA
from webob import Response
from xblock.core import List, Scope, String, XBlock
from xblock.fields import Boolean, Float, Integer
from xblock.validation import ValidationMessage
-from web_fragments.fragment import Fragment
from xblockutils.resources import ResourceLoader
from xblockutils.studio_editable import StudioEditableXBlockMixin
from .exceptions import LtiError
-from .lti import LtiConsumer
+from .lti_1p1.consumer import LtiConsumer1p1, parse_result_json, LTI_PARAMETERS
+from .lti_1p1.oauth import log_authorization_header
from .lti_1p3.exceptions import (
Lti1p3Exception,
UnsupportedGrantType,
@@ -83,7 +82,6 @@ from .lti_1p3.exceptions import (
UnknownClientId,
)
from .lti_1p3.consumer import LtiConsumer1p3
-from .oauth import log_authorization_header
from .outcomes import OutcomeService
from .utils import (
_,
@@ -111,38 +109,6 @@ ROLE_MAP = {
'staff': u'Administrator',
'instructor': u'Instructor',
}
-LTI_PARAMETERS = [
- 'lti_message_type',
- 'lti_version',
- 'resource_link_title',
- 'resource_link_description',
- 'user_image',
- 'lis_person_name_given',
- 'lis_person_name_family',
- 'lis_person_name_full',
- 'lis_person_contact_email_primary',
- 'lis_person_sourcedid',
- 'role_scope_mentor',
- 'context_type',
- 'context_title',
- 'context_label',
- 'launch_presentation_locale',
- 'launch_presentation_document_target',
- 'launch_presentation_css_url',
- 'launch_presentation_width',
- 'launch_presentation_height',
- 'launch_presentation_return_url',
- 'tool_consumer_info_product_family_code',
- 'tool_consumer_info_version',
- 'tool_consumer_instance_guid',
- 'tool_consumer_instance_name',
- 'tool_consumer_instance_description',
- 'tool_consumer_instance_url',
- 'tool_consumer_instance_contact_email',
- 'custom_component_due_date',
- 'custom_component_graceperiod',
- 'custom_component_display_name'
-]
def parse_handler_suffix(suffix):
@@ -580,7 +546,7 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
def validate_field_data(self, validation, data):
if not isinstance(data.custom_parameters, list):
_ = self.runtime.service(self, "i18n").ugettext
- validation.add(ValidationMessage(ValidationMessage.ERROR, six.text_type(
+ validation.add(ValidationMessage(ValidationMessage.ERROR, str(
_("Custom Parameters must be a list")
)))
@@ -668,7 +634,7 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
context_id is an opaque identifier that uniquely identifies the context (e.g., a course)
that contains the link being launched.
"""
- return six.text_type(self.course_id) # pylint: disable=no-member
+ return str(self.course_id) # pylint: disable=no-member
@property
def role(self):
@@ -710,7 +676,7 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
user_id = self.runtime.anonymous_student_id
if user_id is None:
raise LtiError(self.ugettext("Could not get user id for current request"))
- return six.text_type(six.moves.urllib.parse.quote(user_id))
+ return str(urllib.parse.quote(user_id))
def get_icon_class(self):
""" Returns the icon class """
@@ -726,7 +692,7 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
user_id = self.runtime.service(self, 'user').get_external_user_id('lti')
if user_id is None:
raise LtiError(self.ugettext("Could not get user id for current request"))
- return six.text_type(user_id)
+ return str(user_id)
@property
def resource_link_id(self):
@@ -760,7 +726,7 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
i4x-2-3-lti-31de800015cf4afb973356dbe81496df this part of resource_link_id:
makes resource_link_id to be unique among courses inside same system.
"""
- return six.text_type(six.moves.urllib.parse.quote(
+ return str(urllib.parse.quote(
"{}-{}".format(self.runtime.hostname, self.location.html_id()) # pylint: disable=no-member
))
@@ -775,7 +741,7 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
This field is generally optional, but is required for grading.
"""
return "{context}:{resource_link}:{user_id}".format(
- context=six.moves.urllib.parse.quote(self.context_id),
+ context=urllib.parse.quote(self.context_id),
resource_link=self.resource_link_id,
user_id=self.user_id
)
@@ -837,7 +803,19 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
if param_name not in LTI_PARAMETERS:
param_name = 'custom_' + param_name
- custom_parameters[six.text_type(param_name)] = six.text_type(param_value)
+ custom_parameters[param_name] = param_value
+
+ custom_parameters['custom_component_display_name'] = str(self.display_name)
+
+ if self.due: # pylint: disable=no-member
+ custom_parameters.update({
+ 'custom_component_due_date': self.due.strftime('%Y-%m-%d %H:%M:%S') # pylint: disable=no-member
+ })
+ if self.graceperiod: # pylint: disable=no-member
+ custom_parameters.update({
+ 'custom_component_graceperiod': str(self.graceperiod.total_seconds()) # pylint: disable=no-member
+ })
+
return custom_parameters
@property
@@ -852,6 +830,17 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
close_date = due_date
return close_date is not None and timezone.now() > close_date
+ def _get_lti1p1_consumer(self):
+ """
+ Returns a preconfigured LTI 1.1 consumer.
+
+ If the block is configured to use LTI 1.1, set up a
+ base LTI 1.1 consumer class.
+ This class does NOT store state between calls.
+ """
+ key, secret = self.lti_provider_key_secret
+ return LtiConsumer1p1(self.launch_url, key, secret)
+
def _get_lti1p3_consumer(self):
"""
Returns a preconfigured LTI 1.3 consumer.
@@ -876,6 +865,29 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
tool_keyset_url=None,
)
+ def extract_real_user_data(self):
+ """
+ Extract and return real user data from the runtime
+ """
+ user_data = {
+ 'user_email': None,
+ 'user_username': None,
+ 'user_language': None,
+ }
+
+ if callable(self.runtime.get_real_user):
+ real_user_object = self.runtime.get_real_user(self.runtime.anonymous_student_id)
+ user_data['user_email'] = getattr(real_user_object, "email", "")
+ user_data['user_username'] = getattr(real_user_object, "username", "")
+ user_preferences = getattr(real_user_object, "preferences", None)
+
+ if user_preferences is not None:
+ language_preference = user_preferences.filter(key='pref-lang')
+ if len(language_preference) == 1:
+ user_data['user_language'] = language_preference[0].value
+
+ return user_data
+
def studio_view(self, context):
"""
Get Studio View fragment
@@ -976,8 +988,39 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
Returns:
webob.response: HTML LTI launch form
"""
- lti_consumer = LtiConsumer(self)
- lti_parameters = lti_consumer.get_signed_lti_parameters()
+ real_user_data = self.extract_real_user_data()
+
+ lti_consumer = self._get_lti1p1_consumer()
+
+ username = None
+ email = None
+ if self.ask_to_send_username and real_user_data['user_username']:
+ username = real_user_data['user_username']
+ if self.ask_to_send_email and real_user_data['user_email']:
+ email = real_user_data['user_email']
+
+ lti_consumer.set_user_data(
+ self.user_id,
+ self.role,
+ result_sourcedid=self.lis_result_sourcedid,
+ person_sourcedid=username,
+ person_contact_email_primary=email
+ )
+ lti_consumer.set_context_data(
+ self.context_id,
+ self.course.display_name_with_default,
+ self.course.display_org_with_default
+ )
+
+ if self.has_score:
+ lti_consumer.set_outcome_service_url(self.outcome_service_url)
+
+ if real_user_data['user_language']:
+ lti_consumer.set_launch_presentation_locale(real_user_data['user_language'])
+
+ lti_consumer.set_custom_parameters(self.prefixed_custom_parameters)
+
+ lti_parameters = lti_consumer.generate_launch_request(self.resource_link_id)
loader = ResourceLoader(__name__)
context = self._get_context_for_template()
context.update({'lti_parameters': lti_parameters})
@@ -1001,8 +1044,8 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
lti_consumer = self._get_lti1p3_consumer()
context = lti_consumer.prepare_preflight_url(
callback_url=get_lms_lti_launch_link(),
- hint=six.text_type(self.location), # pylint: disable=no-member
- lti_hint=six.text_type(self.location) # pylint: disable=no-member
+ hint=str(self.location), # pylint: disable=no-member
+ lti_hint=str(self.location) # pylint: disable=no-member
)
loader = ResourceLoader(__name__)
@@ -1094,7 +1137,7 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
lti_consumer = self._get_lti1p3_consumer()
try:
token = lti_consumer.access_token(
- dict(parse.parse_qsl(
+ dict(urllib.parse.parse_qsl(
request.body.decode('utf-8'),
keep_blank_values=True
))
@@ -1180,7 +1223,8 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
Returns:
webob.response: response to this request. See above for details.
"""
- lti_consumer = LtiConsumer(self)
+ lti_consumer = self._get_lti1p1_consumer()
+ lti_consumer.set_outcome_service_url(self.outcome_service_url)
if self.runtime.debug:
lti_provider_key, lti_provider_secret = self.lti_provider_key_secret
@@ -1205,21 +1249,83 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
return Response(status=404) # have to do 404 due to spec, but 400 is better, with error msg in body
try:
- # Call the appropriate LtiConsumer method
- args = []
+ # Call the appropriate LtiConsumer1p1 method
+ args = [lti_consumer, user]
if request.method == 'PUT':
# Request body should be passed as an argument
# to result handler method on PUT
args.append(request.body)
- response_body = getattr(lti_consumer, "{}_result".format(request.method.lower()))(user, *args)
+ response_body = getattr(
+ self,
+ "_result_service_{}".format(request.method.lower())
+ )(*args)
except (AttributeError, LtiError):
return Response(status=404)
return Response(
json_body=response_body,
- content_type=LtiConsumer.CONTENT_TYPE_RESULT_JSON,
+ content_type=LtiConsumer1p1.CONTENT_TYPE_RESULT_JSON,
)
+ def _result_service_get(self, lti_consumer, user):
+ """
+ Helper request handler for GET requests to LTI 2.0 result endpoint
+
+ GET handler for lti_2_0_result. Assumes all authorization has been checked.
+
+ Arguments:
+ lti_consumer (lti_consumer.lti_1p1.LtiConsumer1p1): LtiConsumer object that manages Lti1.1 interaction
+ user (django.contrib.auth.models.User): Actual user linked to anon_id in request path suffix
+
+ Returns:
+ dict: response to this request as dictated by the LtiConsumer
+ """
+ self.runtime.rebind_noauth_module_to_user(self, user)
+ args = []
+ if self.module_score:
+ args.extend([self.module_score, self.score_comment])
+ return lti_consumer.get_result(*args)
+
+ def _result_service_delete(self, lti_consumer, user):
+ """
+ Helper request handler for DELETE requests to LTI 2.0 result endpoint
+
+ DELETE handler for lti_2_0_result. Assumes all authorization has been checked.
+
+ Arguments:
+ lti_consumer (lti_consumer.lti_1p1.LtiConsumer1p1): LtiConsumer object that manages Lti1.1 interaction
+ user (django.contrib.auth.models.User): Actual user linked to anon_id in request path suffix
+
+ Returns:
+ dict: response to this request as dictated by the LtiConsumer
+ """
+ self.clear_user_module_score(user)
+ return lti_consumer.delete_result()
+
+ def _result_service_put(self, lti_consumer, user, result_json):
+ """
+ Helper request handler for PUT requests to LTI 2.0 result endpoint
+
+ PUT handler for lti_2_0_result. Assumes all authorization has been checked.
+
+ Arguments:
+ lti_consumer (lti_consumer.lti_1p1.LtiConsumer1p1): LtiConsumer object that manages Lti1.1 interaction
+ request (xblock.django.request.DjangoWebobRequest): Request object
+ real_user (django.contrib.auth.models.User): Actual user linked to anon_id in request path suffix
+
+ Returns:
+ dict: response to this request as dictated by the LtiConsumer
+ """
+ score, comment = parse_result_json(result_json)
+
+ if score is None:
+ # According to http://www.imsglobal.org/lti/ltiv2p0/ltiIMGv2p0.html#_Toc361225514
+ # PUTting a JSON object with no "resultScore" field is equivalent to a DELETE.
+ self.clear_user_module_score(user)
+ else:
+ self.set_user_module_score(user, score, self.max_score(), comment)
+ return lti_consumer.put_result()
+
def max_score(self):
"""
Returns the configured number of possible points for this component.
diff --git a/lti_consumer/outcomes.py b/lti_consumer/outcomes.py
index ef1e347..f20f3dd 100644
--- a/lti_consumer/outcomes.py
+++ b/lti_consumer/outcomes.py
@@ -6,16 +6,14 @@ https://www.imsglobal.org/specs/ltiomv1p0
"""
import logging
+import urllib.parse
from xml.sax.saxutils import escape
-import six.moves.urllib.error
-import six.moves.urllib.parse
from lxml import etree
-from six import text_type
from xblockutils.resources import ResourceLoader
from .exceptions import LtiError
-from .oauth import verify_oauth_body_signature
+from .lti_1p1.oauth import verify_oauth_body_signature
log = logging.getLogger(__name__)
@@ -41,7 +39,7 @@ def parse_grade_xml_body(body):
lti_spec_namespace = "http://www.imsglobal.org/services/ltiv1p1/xsd/imsoms_v1p0"
namespaces = {'def': lti_spec_namespace}
data = body.strip()
- if isinstance(body, text_type):
+ if isinstance(body, str):
data = body.strip().encode('utf-8')
try:
@@ -185,7 +183,7 @@ class OutcomeService(object): # pylint: disable=bad-option-value, useless-objec
log.debug("[LTI]: %s", error_message)
return response_xml_template.format(**failure_values)
- real_user = self.xblock.runtime.get_real_user(six.moves.urllib.parse.unquote(sourced_id.split(':')[-1]))
+ real_user = self.xblock.runtime.get_real_user(urllib.parse.unquote(sourced_id.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_message_identifier)
failure_values['imsx_description'] = "User not found."
diff --git a/lti_consumer/tests/unit/test_lti.py b/lti_consumer/tests/unit/test_lti.py
deleted file mode 100644
index 9128b6f..0000000
--- a/lti_consumer/tests/unit/test_lti.py
+++ /dev/null
@@ -1,349 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-Unit tests for lti_consumer.lti module
-"""
-
-import unittest
-from datetime import timedelta
-
-import six
-from django.utils import timezone
-from mock import Mock, PropertyMock, patch
-from six import text_type
-
-from lti_consumer.exceptions import LtiError
-from lti_consumer.lti import LtiConsumer, parse_result_json
-from lti_consumer.tests.unit.test_lti_consumer import TestLtiConsumerXBlock
-from lti_consumer.tests.unit.test_utils import (make_request,
- patch_signed_parameters)
-
-INVALID_JSON_INPUTS = [
- ([
- u"kk", # ValueError
- u"{{}", # ValueError
- u"{}}", # ValueError
- 3, # TypeError
- {}, # TypeError
- ], u"Supplied JSON string in request body could not be decoded"),
- ([
- u"3", # valid json, not array or object
- u"[]", # valid json, array too small
- u"[3, {}]", # valid json, 1st element not an object
- ], u"Supplied JSON string is a list that does not contain an object as the first element"),
- ([
- u'{"@type": "NOTResult"}', # @type key must have value 'Result'
- ], u"JSON object does not contain correct @type attribute"),
- ([
- # @context missing
- u'{"@type": "Result", "resultScore": 0.1}',
- ], u"JSON object does not contain required key"),
- ([
- u'''
- {"@type": "Result",
- "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
- "resultScore": 100}''' # score out of range
- ], u"score value outside the permitted range of 0.0-1.0."),
- ([
- u'''
- {"@type": "Result",
- "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
- "resultScore": -2}''' # score out of range
- ], u"score value outside the permitted range of 0.0-1.0."),
- ([
- u'''
- {"@type": "Result",
- "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
- "resultScore": "1b"}''', # score ValueError
- u'''
- {"@type": "Result",
- "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
- "resultScore": {}}''', # score TypeError
- ], u"Could not convert resultScore to float"),
-]
-
-VALID_JSON_INPUTS = [
- (u'''
- {"@type": "Result",
- "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
- "resultScore": 0.1}''', 0.1, u""), # no comment means we expect ""
- (u'''
- [{"@type": "Result",
- "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
- "@id": "anon_id:abcdef0123456789",
- "resultScore": 0.1}]''', 0.1, u""), # OK to have array of objects -- just take the first. @id is okay too
- (u'''
- {"@type": "Result",
- "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
- "resultScore": 0.1,
- "comment": "ಠ益ಠ"}''', 0.1, u"ಠ益ಠ"), # unicode comment
- (u'''
- {"@type": "Result",
- "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result"}''', None, u""), # no score means we expect None
- (u'''
- {"@type": "Result",
- "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
- "resultScore": 0.0}''', 0.0, u""), # test lower score boundary
- (u'''
- {"@type": "Result",
- "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
- "resultScore": 1.0}''', 1.0, u""), # test upper score boundary
-]
-
-GET_RESULT_RESPONSE = {
- "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
- "@type": "Result",
-}
-
-
-class TestParseResultJson(unittest.TestCase):
- """
- Unit tests for `lti_consumer.lti.parse_result_json`
- """
-
- def test_invalid_json(self):
- """
- Test invalid json raises exception
- """
- for error_inputs, error_message in INVALID_JSON_INPUTS:
- for error_input in error_inputs:
- with six.assertRaisesRegex(self, LtiError, error_message):
- parse_result_json(error_input)
-
- def test_valid_json(self):
- """
- Test valid json returns expected values
- """
- for json_str, expected_score, expected_comment in VALID_JSON_INPUTS:
- score, comment = parse_result_json(json_str)
- self.assertEqual(score, expected_score)
- self.assertEqual(comment, expected_comment)
-
-
-class TestLtiConsumer(TestLtiConsumerXBlock):
- """
- Unit tests for LtiConsumer
- """
-
- def setUp(self):
- super(TestLtiConsumer, self).setUp()
- self.lti_consumer = LtiConsumer(self.xblock)
-
- def _update_xblock_for_signed_parameters(self):
- """
- Prepare the LTI XBlock for signing the parameters.
- """
- self.lti_consumer.xblock.due = timezone.now()
- self.lti_consumer.xblock.graceperiod = timedelta(days=1)
- self.lti_consumer.xblock.has_score = True
- self.lti_consumer.xblock.ask_to_send_username = True
- self.lti_consumer.xblock.ask_to_send_email = True
- self.lti_consumer.xblock.runtime.get_real_user.return_value = Mock(
- email='edx@example.com',
- username='edx',
- preferences=Mock(filter=Mock(return_value=[Mock(value='en')]))
- )
-
- @patch_signed_parameters
- def test_get_signed_lti_parameters(self):
- """
- Test `get_signed_lti_parameters` returns the correct dict
- """
- self._update_xblock_for_signed_parameters()
- expected_lti_parameters = {
- text_type('user_id'): self.lti_consumer.xblock.user_id,
- text_type('oauth_callback'): 'about:blank',
- text_type('launch_presentation_return_url'): '',
- text_type('lti_message_type'): 'basic-lti-launch-request',
- text_type('lti_version'): 'LTI-1p0',
- text_type('roles'): self.lti_consumer.xblock.role,
- text_type('resource_link_id'): self.lti_consumer.xblock.resource_link_id,
- text_type('lis_result_sourcedid'): self.lti_consumer.xblock.lis_result_sourcedid,
- text_type('context_id'): self.lti_consumer.xblock.context_id,
- text_type('lis_outcome_service_url'): self.lti_consumer.xblock.outcome_service_url,
- text_type('custom_component_display_name'): self.lti_consumer.xblock.display_name,
- text_type('custom_component_due_date'): self.lti_consumer.xblock.due.strftime('%Y-%m-%d %H:%M:%S'),
- text_type('custom_component_graceperiod'): str(self.lti_consumer.xblock.graceperiod.total_seconds()),
- 'lis_person_sourcedid': 'edx',
- 'lis_person_contact_email_primary': 'edx@example.com',
- 'launch_presentation_locale': 'en',
- text_type('custom_param_1'): 'custom1',
- text_type('custom_param_2'): 'custom2',
- text_type('oauth_nonce'): 'fake_nonce',
- 'oauth_timestamp': 'fake_timestamp',
- 'oauth_version': 'fake_version',
- 'oauth_signature_method': 'fake_method',
- 'oauth_consumer_key': 'fake_consumer_key',
- 'oauth_signature': 'fake_signature',
- text_type('context_label'): self.lti_consumer.xblock.course.display_org_with_default,
- text_type('context_title'): self.lti_consumer.xblock.course.display_name_with_default,
- }
- self.assertEqual(self.lti_consumer.get_signed_lti_parameters(), expected_lti_parameters)
-
- # Test that `lis_person_sourcedid`, `lis_person_contact_email_primary`, and `launch_presentation_locale`
- # are not included in the returned LTI parameters when a user cannot be found
- self.lti_consumer.xblock.runtime.get_real_user.return_value = {}
- del expected_lti_parameters['lis_person_sourcedid']
- del expected_lti_parameters['lis_person_contact_email_primary']
- del expected_lti_parameters['launch_presentation_locale']
- self.assertEqual(self.lti_consumer.get_signed_lti_parameters(), expected_lti_parameters)
-
- @patch_signed_parameters
- @patch('lti_consumer.lti.log')
- def test_parameter_processors(self, mock_log):
- self._update_xblock_for_signed_parameters()
- self.xblock.enable_processors = True
-
- mock_value = {
- 'parameter_processors': ['lti_consumer.tests.unit.test_utils:dummy_processor']
- }
- with patch('lti_consumer.lti_xblock.LtiConsumerXBlock.get_settings', return_value=mock_value):
- params = self.lti_consumer.get_signed_lti_parameters()
- assert params['custom_author_country'] == u''
- assert params['custom_author_email'] == u'author@example.com'
- assert not mock_log.exception.called
-
- @patch_signed_parameters
- @patch('lti_consumer.lti.log')
- def test_default_params(self, mock_log):
- self._update_xblock_for_signed_parameters()
- self.xblock.enable_processors = True
-
- mock_value = {
- 'parameter_processors': ['lti_consumer.tests.unit.test_utils:defaulting_processor']
- }
- with patch('lti_consumer.lti_xblock.LtiConsumerXBlock.get_settings', return_value=mock_value):
- params = self.lti_consumer.get_signed_lti_parameters()
- assert params['custom_country'] == u''
- assert params['custom_name'] == u'Lex'
- assert not mock_log.exception.called
-
- @patch_signed_parameters
- @patch('lti_consumer.lti.log')
- def test_default_params_with_error(self, mock_log):
- self._update_xblock_for_signed_parameters()
- self.xblock.enable_processors = True
-
- mock_value = {
- 'parameter_processors': ['lti_consumer.tests.unit.test_utils:faulty_processor']
- }
- with patch('lti_consumer.lti_xblock.LtiConsumerXBlock.get_settings', return_value=mock_value):
- params = self.lti_consumer.get_signed_lti_parameters()
- assert params['custom_name'] == u'Lex'
- assert mock_log.exception.called
-
- def test_get_result(self):
- """
- Test `get_result` returns valid json response
- """
- self.xblock.module_score = 0.9
- self.xblock.score_comment = 'Great Job!'
- response = dict(GET_RESULT_RESPONSE)
- response.update({
- "resultScore": self.xblock.module_score,
- "comment": self.xblock.score_comment
- })
- self.assertEqual(self.lti_consumer.get_result(Mock()), response)
-
- self.xblock.module_score = None
- self.xblock.score_comment = ''
- self.assertEqual(self.lti_consumer.get_result(Mock()), GET_RESULT_RESPONSE)
-
- @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.clear_user_module_score')
- def test_delete_result(self, mock_clear):
- """
- Test `delete_result` calls `LtiConsumerXBlock.clear_user_module_score`
- """
- user = Mock()
- response = self.lti_consumer.delete_result(user)
-
- mock_clear.assert_called_with(user)
- self.assertEqual(response, {})
-
- @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.max_score', Mock(return_value=1.0))
- @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.set_user_module_score')
- @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.clear_user_module_score')
- @patch('lti_consumer.lti.parse_result_json')
- def test_put_result(self, mock_parse, mock_clear, mock_set):
- """
- Test `put_result` calls `LtiConsumerXBlock.set_user_module_score`
- or `LtiConsumerXblock.clear_user_module_score` if resultScore not included in request
- """
- user = Mock()
- score = 0.9
- comment = 'Great Job!'
-
- mock_parse.return_value = (score, comment)
- response = self.lti_consumer.put_result(user, '')
- mock_set.assert_called_with(user, score, 1.0, comment)
- self.assertEqual(response, {})
-
- mock_parse.return_value = (None, '')
- response = self.lti_consumer.put_result(user, '')
- mock_clear.assert_called_with(user)
- self.assertEqual(response, {})
-
- @patch('lti_consumer.lti.log')
- def test_verify_result_headers_verify_content_type_true(self, mock_log):
- """
- Test wrong content type raises exception if `verify_content_type` is True
- """
- request = make_request('')
-
- with self.assertRaises(LtiError):
- self.lti_consumer.verify_result_headers(request)
-
- assert mock_log.error.called
-
- @patch('lti_consumer.lti.verify_oauth_body_signature', Mock(return_value=True))
- @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.lti_provider_key_secret', PropertyMock(return_value=('t', 's')))
- def test_verify_result_headers_verify_content_type_false(self):
- """
- Test content type check skipped if `verify_content_type` is False
- """
- request = make_request('')
- request.environ['CONTENT_TYPE'] = LtiConsumer.CONTENT_TYPE_RESULT_JSON
- response = self.lti_consumer.verify_result_headers(request, False)
-
- self.assertTrue(response)
-
- @patch('lti_consumer.lti.verify_oauth_body_signature', Mock(return_value=True))
- @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.lti_provider_key_secret', PropertyMock(return_value=('t', 's')))
- def test_verify_result_headers_valid(self):
- """
- Test True is returned if request is valid
- """
- request = make_request('')
- request.environ['CONTENT_TYPE'] = LtiConsumer.CONTENT_TYPE_RESULT_JSON
- response = self.lti_consumer.verify_result_headers(request)
-
- self.assertTrue(response)
-
- @patch('lti_consumer.lti.verify_oauth_body_signature', Mock(side_effect=LtiError))
- @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.lti_provider_key_secret', PropertyMock(return_value=('t', 's')))
- @patch('lti_consumer.lti.log')
- def test_verify_result_headers_lti_error(self, mock_log):
- """
- Test exception raised if request header verification raises error
- """
- request = make_request('')
- request.environ['CONTENT_TYPE'] = LtiConsumer.CONTENT_TYPE_RESULT_JSON
-
- with self.assertRaises(LtiError):
- self.lti_consumer.verify_result_headers(request)
-
- assert mock_log.error.called
-
- @patch('lti_consumer.lti.verify_oauth_body_signature', Mock(side_effect=ValueError))
- @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.lti_provider_key_secret', PropertyMock(return_value=('t', 's')))
- @patch('lti_consumer.lti.log')
- def test_verify_result_headers_value_error(self, mock_log):
- """
- Test exception raised if request header verification raises error
- """
- request = make_request('')
- request.environ['CONTENT_TYPE'] = LtiConsumer.CONTENT_TYPE_RESULT_JSON
-
- with self.assertRaises(LtiError):
- self.lti_consumer.verify_result_headers(request)
-
- assert mock_log.error.called
diff --git a/lti_consumer/tests/unit/test_lti_consumer.py b/lti_consumer/tests/unit/test_lti_consumer.py
index ea37658..4c4a913 100644
--- a/lti_consumer/tests/unit/test_lti_consumer.py
+++ b/lti_consumer/tests/unit/test_lti_consumer.py
@@ -4,16 +4,15 @@ Unit tests for LtiConsumerXBlock
from datetime import timedelta
import json
+import urllib.parse
import uuid
import ddt
-import six
-from six.moves.urllib import parse
from Crypto.PublicKey import RSA
from django.test.testcases import TestCase
from django.utils import timezone
from jwkest.jwk import RSAKey
-from mock import Mock, PropertyMock, patch
+from mock import Mock, PropertyMock, NonCallableMock, patch
from lti_consumer.exceptions import LtiError
from lti_consumer.lti_xblock import LtiConsumerXBlock, parse_handler_suffix
@@ -93,7 +92,7 @@ class TestProperties(TestLtiConsumerXBlock):
"""
Test `context_id` returns unicode course id
"""
- self.assertEqual(self.xblock.context_id, six.text_type(self.xblock.course_id)) # pylint: disable=no-member
+ self.assertEqual(self.xblock.context_id, str(self.xblock.course_id)) # pylint: disable=no-member
def test_validate(self):
"""
@@ -243,10 +242,25 @@ class TestProperties(TestLtiConsumerXBlock):
"""
Test `prefixed_custom_parameters` appropriately prefixes the configured custom params
"""
+ now = timezone.now()
+ one_day = timedelta(days=1)
+ self.xblock.due = now
+ self.xblock.graceperiod = one_day
+
self.xblock.custom_parameters = ['param_1=true', 'param_2 = false', 'lti_version=1.1']
+
+ expected_params = {
+ u'custom_component_display_name': self.xblock.display_name,
+ u'custom_component_due_date': now.strftime('%Y-%m-%d %H:%M:%S'),
+ u'custom_component_graceperiod': str(one_day.total_seconds()),
+ u'custom_param_1': u'true',
+ u'custom_param_2': u'false',
+ u'lti_version': u'1.1'
+ }
+
params = self.xblock.prefixed_custom_parameters
- self.assertEqual(params, {u'custom_param_1': u'true', u'custom_param_2': u'false', u'lti_version': u'1.1'})
+ self.assertEqual(params, expected_params)
def test_invalid_custom_parameter(self):
"""
@@ -386,6 +400,76 @@ class TestEditableFields(TestLtiConsumerXBlock):
lti_1p3_enabled_mock.assert_called()
+class TestGetLti1p1Consumer(TestLtiConsumerXBlock):
+ """
+ Unit tests for LtiConsumerXBlock._get_lti1p1_consumer()
+ """
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.course')
+ @patch('lti_consumer.lti_xblock.LtiConsumer1p1')
+ def test_lti_1p1_consumer_created(self, mock_lti_consumer, mock_course):
+ """
+ Test LtiConsumer1p1 is created with the launch_url, oauth_key, and oauth_secret
+ """
+ provider = 'lti_provider'
+ key = 'test'
+ secret = 'secret'
+ self.xblock.lti_id = provider
+ type(mock_course).lti_passports = PropertyMock(return_value=["{}:{}:{}".format(provider, key, secret)])
+
+ self.xblock._get_lti1p1_consumer() # pylint: disable=protected-access
+
+ mock_lti_consumer.assert_called_with(self.xblock.launch_url, key, secret)
+
+
+class TestExtractRealUserData(TestLtiConsumerXBlock):
+ """
+ Unit tests for LtiConsumerXBlock.extract_real_user_data()
+ """
+
+ def test_get_real_user_not_callable(self):
+ """
+ Test user_email, user_username, and user_language not available
+ """
+ self.xblock.runtime.get_real_user = NonCallableMock()
+
+ real_user_data = self.xblock.extract_real_user_data()
+ self.assertIsNone(real_user_data['user_email'])
+ self.assertIsNone(real_user_data['user_username'])
+ self.assertIsNone(real_user_data['user_language'])
+
+ def test_get_real_user_callable(self):
+ """
+ Test user_email, and user_username available, but not user_language
+ """
+ fake_user = Mock()
+ fake_user.email = 'abc@example.com'
+ fake_user.username = 'fake'
+ fake_user.preferences = None
+
+ self.xblock.runtime.get_real_user = Mock(return_value=fake_user)
+
+ real_user_data = self.xblock.extract_real_user_data()
+ self.assertEqual(real_user_data['user_email'], fake_user.email)
+ self.assertEqual(real_user_data['user_username'], fake_user.username)
+ self.assertIsNone(real_user_data['user_language'])
+
+ def test_get_real_user_callable_with_language_preference(self):
+ """
+ Test user_language available
+ """
+ fake_user = Mock()
+ fake_user.email = 'abc@example.com'
+ fake_user.username = 'fake'
+ mock_language_pref = Mock()
+ mock_language_pref.value = PropertyMock(return_value='en')
+ fake_user.preferences.filter = Mock(return_value=[mock_language_pref])
+
+ self.xblock.runtime.get_real_user = Mock(return_value=fake_user)
+
+ real_user_data = self.xblock.extract_real_user_data()
+ self.assertEqual(real_user_data['user_language'], mock_language_pref.value)
+
+
class TestStudentView(TestLtiConsumerXBlock):
"""
Unit tests for LtiConsumerXBlock.student_view()
@@ -476,15 +560,29 @@ class TestLtiLaunchHandler(TestLtiConsumerXBlock):
Unit tests for LtiConsumerXBlock.lti_launch_handler()
"""
- @patch('lti_consumer.lti.LtiConsumer.get_signed_lti_parameters')
- def test_handle_request_called(self, mock_get_signed_lti_parameters):
+ def setUp(self):
+ super(TestLtiLaunchHandler, self).setUp()
+ self.mock_lti_consumer = Mock(generate_launch_request=Mock(return_value={}))
+ self.xblock._get_lti1p1_consumer = Mock(return_value=self.mock_lti_consumer) # pylint: disable=protected-access
+ self.xblock.due = timezone.now()
+ self.xblock.graceperiod = timedelta(days=1)
+ self.xblock.runtime.get_real_user = Mock(return_value=None)
+
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.course')
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.user_id', PropertyMock(return_value=FAKE_USER_ID))
+ def test_generate_launch_request_called(self, mock_course):
"""
- Test LtiConsumer.get_signed_lti_parameters is called and a 200 HTML response is returned
+ Test LtiConsumer.generate_launch_request is called and a 200 HTML response is returned
"""
+ provider = 'lti_provider'
+ key = 'test'
+ secret = 'secret'
+ type(mock_course).lti_passports = PropertyMock(return_value=["{}:{}:{}".format(provider, key, secret)])
+
request = make_request('', 'GET')
response = self.xblock.lti_launch_handler(request)
- assert mock_get_signed_lti_parameters.called
+ self.mock_lti_consumer.generate_launch_request.assert_called_with(self.xblock.resource_link_id)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.content_type, 'text/html')
@@ -519,6 +617,8 @@ class TestResultServiceHandler(TestLtiConsumerXBlock):
self.xblock.runtime.debug = False
self.xblock.runtime.get_real_user = Mock()
self.xblock.accept_grades_past_due = True
+ self.mock_lti_consumer = Mock()
+ self.xblock._get_lti1p1_consumer = Mock(return_value=self.mock_lti_consumer) # pylint: disable=protected-access
@patch('lti_consumer.lti_xblock.log_authorization_header')
@patch('lti_consumer.lti_xblock.LtiConsumerXBlock.lti_provider_key_secret')
@@ -555,18 +655,16 @@ class TestResultServiceHandler(TestLtiConsumerXBlock):
self.assertEqual(response.status_code, 404)
- @patch('lti_consumer.lti.LtiConsumer.get_result')
- @patch('lti_consumer.lti.LtiConsumer.verify_result_headers', Mock(return_value=True))
+ @patch('lti_consumer.lti_1p1.consumer.LtiConsumer1p1.verify_result_headers', Mock(return_value=True))
@patch('lti_consumer.lti_xblock.parse_handler_suffix')
@patch('lti_consumer.lti_xblock.LtiConsumerXBlock.is_past_due')
- def test_accept_grades_past_due_true_and_is_past_due_true(self, mock_is_past_due, mock_parse_suffix,
- mock_get_result):
+ def test_accept_grades_past_due_true_and_is_past_due_true(self, mock_is_past_due, mock_parse_suffix):
"""
Test 200 response returned when `accept_grades_past_due` is True and `is_past_due` is True
"""
mock_is_past_due.__get__ = Mock(return_value=True)
mock_parse_suffix.return_value = FAKE_USER_ID
- mock_get_result.return_value = {}
+ self.mock_lti_consumer.get_result.return_value = {}
response = self.xblock.result_service_handler(make_request('', 'GET'))
self.assertEqual(response.status_code, 200)
@@ -581,19 +679,18 @@ class TestResultServiceHandler(TestLtiConsumerXBlock):
self.assertEqual(response.status_code, 404)
- @patch('lti_consumer.lti.LtiConsumer.verify_result_headers')
@patch('lti_consumer.lti_xblock.parse_handler_suffix')
- def test_verify_headers_raises_error(self, mock_parse_suffix, mock_verify_result_headers):
+ def test_verify_headers_raises_error(self, mock_parse_suffix):
"""
Test 401 response returned when `verify_result_headers` raises LtiError
"""
mock_parse_suffix.return_value = FAKE_USER_ID
- mock_verify_result_headers.side_effect = LtiError()
+ self.mock_lti_consumer.verify_result_headers.side_effect = LtiError()
response = self.xblock.result_service_handler(make_request('', 'GET'))
self.assertEqual(response.status_code, 401)
- @patch('lti_consumer.lti.LtiConsumer.verify_result_headers', Mock(return_value=True))
+ @patch('lti_consumer.lti_1p1.consumer.LtiConsumer1p1.verify_result_headers', Mock(return_value=True))
@patch('lti_consumer.lti_xblock.parse_handler_suffix')
def test_bad_user_id(self, mock_parse_suffix):
"""
@@ -605,7 +702,7 @@ class TestResultServiceHandler(TestLtiConsumerXBlock):
self.assertEqual(response.status_code, 404)
- @patch('lti_consumer.lti.LtiConsumer.verify_result_headers', Mock(return_value=True))
+ @patch('lti_consumer.lti_1p1.consumer.LtiConsumer1p1.verify_result_headers', Mock(return_value=True))
@patch('lti_consumer.lti_xblock.parse_handler_suffix')
def test_bad_request_method(self, mock_parse_suffix):
"""
@@ -616,61 +713,139 @@ class TestResultServiceHandler(TestLtiConsumerXBlock):
self.assertEqual(response.status_code, 404)
- @patch('lti_consumer.lti.LtiConsumer.get_result')
- @patch('lti_consumer.lti.LtiConsumer.verify_result_headers', Mock(return_value=True))
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock._result_service_get')
+ @patch('lti_consumer.lti_1p1.consumer.LtiConsumer1p1.verify_result_headers', Mock(return_value=True))
@patch('lti_consumer.lti_xblock.parse_handler_suffix')
- def test_get_result_raises_error(self, mock_parse_suffix, mock_get_result):
+ def test_get_result_raises_error(self, mock_parse_suffix, mock_result_service_get):
"""
- Test 404 response returned when the LtiConsumer result service handler methods raise an exception
+ Test 404 response returned when the LtiConsumerXBlock._result_service_* methods raise an exception
"""
mock_parse_suffix.return_value = FAKE_USER_ID
- mock_get_result.side_effect = LtiError()
+ mock_result_service_get.side_effect = LtiError()
response = self.xblock.result_service_handler(make_request('', 'GET'))
self.assertEqual(response.status_code, 404)
- @patch('lti_consumer.lti.LtiConsumer.get_result')
- @patch('lti_consumer.lti.LtiConsumer.verify_result_headers', Mock(return_value=True))
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock._result_service_get')
+ @patch('lti_consumer.lti_1p1.consumer.LtiConsumer1p1.verify_result_headers', Mock(return_value=True))
@patch('lti_consumer.lti_xblock.parse_handler_suffix')
- def test_get_result_called(self, mock_parse_suffix, mock_get_result):
+ def test_result_service_get_called(self, mock_parse_suffix, mock_result_service_get):
"""
- Test 200 response and LtiConsumer.get_result is called on a GET request
+ Test 200 response and LtiConsumerXBlock._result_service_get is called on a GET request
"""
mock_parse_suffix.return_value = FAKE_USER_ID
- mock_get_result.return_value = {}
+ mock_result_service_get.return_value = {}
response = self.xblock.result_service_handler(make_request('', 'GET'))
- assert mock_get_result.called
+ assert mock_result_service_get.called
self.assertEqual(response.status_code, 200)
- @patch('lti_consumer.lti.LtiConsumer.put_result')
- @patch('lti_consumer.lti.LtiConsumer.verify_result_headers', Mock(return_value=True))
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock._result_service_put')
+ @patch('lti_consumer.lti_1p1.consumer.LtiConsumer1p1.verify_result_headers', Mock(return_value=True))
@patch('lti_consumer.lti_xblock.parse_handler_suffix')
- def test_put_result_called(self, mock_parse_suffix, mock_put_result):
+ def test_result_service_put_called(self, mock_parse_suffix, mock_result_service_put):
"""
- Test 200 response and LtiConsumer.put_result is called on a PUT request
+ Test 200 response and LtiConsumerXBlock._result_service_put is called on a PUT request
"""
mock_parse_suffix.return_value = FAKE_USER_ID
- mock_put_result.return_value = {}
+ mock_result_service_put.return_value = {}
response = self.xblock.result_service_handler(make_request('', 'PUT'))
- assert mock_put_result.called
+ assert mock_result_service_put.called
self.assertEqual(response.status_code, 200)
- @patch('lti_consumer.lti.LtiConsumer.delete_result')
- @patch('lti_consumer.lti.LtiConsumer.verify_result_headers', Mock(return_value=True))
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock._result_service_delete')
+ @patch('lti_consumer.lti_1p1.consumer.LtiConsumer1p1.verify_result_headers', Mock(return_value=True))
@patch('lti_consumer.lti_xblock.parse_handler_suffix')
- def test_delete_result_called(self, mock_parse_suffix, mock_delete_result):
+ def test_result_service_delete_called(self, mock_parse_suffix, mock_result_service_delete):
"""
- Test 200 response and LtiConsumer.delete_result is called on a DELETE request
+ Test 200 response and LtiConsumerXBlock._result_service_delete is called on a DELETE request
"""
mock_parse_suffix.return_value = FAKE_USER_ID
- mock_delete_result.return_value = {}
+ mock_result_service_delete.return_value = {}
response = self.xblock.result_service_handler(make_request('', 'DELETE'))
- assert mock_delete_result.called
+ assert mock_result_service_delete.called
self.assertEqual(response.status_code, 200)
+ def test_consumer_get_result_called(self):
+ """
+ Test runtime calls rebind_noauth_module_to_user and LtiConsumer.get_result is called on a GET request
+ """
+ mock_runtime = self.xblock.runtime = Mock()
+ mock_lti_consumer = Mock()
+ mock_user = Mock()
+
+ self.xblock._result_service_get(mock_lti_consumer, mock_user) # pylint: disable=protected-access
+
+ mock_runtime.rebind_noauth_module_to_user.assert_called_with(self.xblock, mock_user)
+ mock_lti_consumer.get_result.assert_called_with()
+
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.module_score', PropertyMock(return_value=0.5))
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.score_comment', PropertyMock(return_value='test'))
+ def test_consumer_get_result_called_with_score_details(self):
+ """
+ Test LtiConsumer.get_result is called with module_score and score_comment on a GET request with a module_score
+ """
+ mock_lti_consumer = Mock()
+ mock_user = Mock()
+
+ self.xblock._result_service_get(mock_lti_consumer, mock_user) # pylint: disable=protected-access
+
+ mock_lti_consumer.get_result.assert_called_with(0.5, 'test')
+
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.clear_user_module_score', Mock(return_value=True))
+ @patch('lti_consumer.lti_xblock.parse_result_json')
+ def test_consumer_put_result_called(self, mock_parse_result_json):
+ """
+ Test parse_result_json and LtiConsumer.put_result is called on a PUT request
+ """
+ mock_parse_result_json.return_value = (None, None)
+ mock_lti_consumer = Mock()
+ mock_user = Mock()
+
+ self.xblock._result_service_put(mock_lti_consumer, mock_user, '') # pylint: disable=protected-access
+
+ assert mock_parse_result_json.called
+ assert mock_lti_consumer.put_result.called
+
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.clear_user_module_score')
+ @patch('lti_consumer.lti_xblock.parse_result_json', Mock(return_value=(None, None)))
+ def test_clear_user_module_score_called_when_no_score_available(self, mock_clear_user_module_score):
+ """
+ Test LtiConsumerXBlock.clear_user_module_score is called on a PUT request with no score
+ """
+ mock_lti_consumer = Mock()
+ mock_user = Mock()
+ self.xblock._result_service_put(mock_lti_consumer, mock_user, '') # pylint: disable=protected-access
+
+ mock_clear_user_module_score.assert_called_with(mock_user)
+
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.set_user_module_score')
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.max_score', Mock(return_value=10))
+ @patch('lti_consumer.lti_xblock.parse_result_json', Mock(return_value=(1, 'comment')))
+ def test_set_user_module_score_called_when_score_available(self, mock_set_user_module_score):
+ """
+ Test LtiConsumerXBlock.set_user_module_score is called on a PUT request with a score
+ """
+ mock_lti_consumer = Mock()
+ mock_user = Mock()
+ self.xblock._result_service_put(mock_lti_consumer, mock_user, '') # pylint: disable=protected-access
+
+ mock_set_user_module_score.assert_called_with(mock_user, 1, 10, 'comment')
+
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.clear_user_module_score')
+ def test_consumer_delete_result_called(self, mock_clear_user_module_score):
+ """
+ Test LtiConsumerXBlock.clear_user_module_score is called on a PUT request with no score
+ """
+ mock_lti_consumer = Mock()
+ mock_user = Mock()
+ self.xblock._result_service_delete(mock_lti_consumer, mock_user) # pylint: disable=protected-access
+
+ mock_clear_user_module_score.assert_called_with(mock_user)
+ assert mock_lti_consumer.delete_result.called
+
def test_get_outcome_service_url_with_default_parameter(self):
"""
Test `get_outcome_service_url` with default parameter
@@ -1195,7 +1370,7 @@ class TestLti1p3AccessTokenEndpoint(TestCase):
Test request with invalid JWT.
"""
request = make_request(
- parse.urlencode({
+ urllib.parse.urlencode({
"grant_type": "client_credentials",
"client_assertion_type": "something",
"client_assertion": "invalid-jwt",
@@ -1214,7 +1389,7 @@ class TestLti1p3AccessTokenEndpoint(TestCase):
Test request with invalid grant.
"""
request = make_request(
- parse.urlencode({
+ urllib.parse.urlencode({
"grant_type": "password",
"client_assertion_type": "something",
"client_assertion": "invalit-jwt",
@@ -1237,7 +1412,7 @@ class TestLti1p3AccessTokenEndpoint(TestCase):
jwt = create_jwt(self.key, {})
request = make_request(
- parse.urlencode({
+ urllib.parse.urlencode({
"grant_type": "client_credentials",
"client_assertion_type": "something",
"client_assertion": jwt,
@@ -1257,7 +1432,7 @@ class TestLti1p3AccessTokenEndpoint(TestCase):
"""
jwt = create_jwt(self.key, {})
request = make_request(
- parse.urlencode({
+ urllib.parse.urlencode({
"grant_type": "client_credentials",
"client_assertion_type": "something",
"client_assertion": jwt,
diff --git a/lti_consumer/tests/unit/test_utils.py b/lti_consumer/tests/unit/test_utils.py
index fdf57de..0ad41da 100644
--- a/lti_consumer/tests/unit/test_utils.py
+++ b/lti_consumer/tests/unit/test_utils.py
@@ -2,7 +2,6 @@
Utility functions used within unit tests
"""
-import six
from mock import Mock, PropertyMock, patch
from webob import Request
from workbench.runtime import WorkbenchRuntime
@@ -29,7 +28,7 @@ def make_xblock(xblock_name, xblock_cls, attributes):
hostname='localhost',
)
xblock.course_id = 'course-v1:edX+DemoX+Demo_Course'
- for key, value in six.iteritems(attributes):
+ for key, value in attributes.items():
setattr(xblock, key, value)
return xblock
diff --git a/requirements/travis.in b/requirements/travis.in
index 5aa07ac..c2bccb5 100644
--- a/requirements/travis.in
+++ b/requirements/travis.in
@@ -3,5 +3,3 @@
-r test.txt
-r tox.txt
-
-six
diff --git a/setup.py b/setup.py
index 15734d9..adf045c 100644
--- a/setup.py
+++ b/setup.py
@@ -49,7 +49,7 @@ with open('README.rst') as _f:
setup(
name='lti-consumer-xblock',
- version='2.0.1.1',
+ version='2.0.2',
description='This XBlock implements the consumer side of the LTI specification.',
long_description=long_description,
long_description_content_type='text/markdown',
--
GitLab