From ba7044e9a77605bc6cbe5c9c7d6e922863483176 Mon Sep 17 00:00:00 2001 From: Patrick Cockwell 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 0d350459..de9cd7e2 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 e80b55f5..dec54692 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 184b2735..00000000 --- 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 00000000..92410b4e --- /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 00000000..e69de29b diff --git a/lti_consumer/lti_1p1/consumer.py b/lti_consumer/lti_1p1/consumer.py new file mode 100644 index 00000000..4903439d --- /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 00000000..e69de29b diff --git a/lti_consumer/lti_1p1/contrib/django.py b/lti_consumer/lti_1p1/contrib/django.py new file mode 100644 index 00000000..137c5e72 --- /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 00000000..e69de29b 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 00000000..c958584e --- /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 00000000..80383ae8 --- /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 8bd5d890..61f8daad 100644 --- a/lti_consumer/oauth.py +++ b/lti_consumer/lti_1p1/oauth.py @@ -5,13 +5,11 @@ 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 00000000..e69de29b 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 00000000..06389b03 --- /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 738d7a63..d7ef35bf 100644 --- a/lti_consumer/tests/unit/test_oauth.py +++ b/lti_consumer/lti_1p1/tests/test_oauth.py @@ -6,10 +6,10 @@ 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 @@ def test_sign_raises_error(self, mock_client_sign): """ mock_client_sign.side_effect = ValueError - with self.assertRaises(LtiError): + with self.assertRaises(Lti1p1Error): __ = get_oauth_request_signature('test', 'secret', '', {}, '') @@ -69,7 +69,7 @@ def test_invalid_signature(self): """ 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 @@ def test_missing_oauth_body_hash(self): """ 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 5384111e..f5b923bf 100644 --- a/lti_consumer/lti_xblock.py +++ b/lti_consumer/lti_xblock.py @@ -50,29 +50,28 @@ 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 @@ UnknownClientId, ) from .lti_1p3.consumer import LtiConsumer1p3 -from .oauth import log_authorization_header from .outcomes import OutcomeService from .utils import ( _, @@ -111,38 +109,6 @@ '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 @@ def workbench_scenarios(): 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 @@ def context_id(self): 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 @@ def user_id(self): 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 @@ def external_user_id(self): 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 @@ def resource_link_id(self): 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 @@ def lis_result_sourcedid(self): 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 @@ def prefixed_custom_parameters(self): 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 @@ def is_past_due(self): 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 @@ def _get_lti1p3_consumer(self): 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 @@ def lti_launch_handler(self, request, suffix=''): # pylint: disable=unused-argu 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 @@ def lti_1p3_launch_handler(self, request, suffix=''): # pylint: disable=unused- 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 @@ def lti_1p3_access_token(self, request, suffix=''): # pylint: disable=unused-ar 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 @@ def result_service_handler(self, request, suffix=''): 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 @@ def result_service_handler(self, request, suffix=''): 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 ef1e3477..f20f3dd5 100644 --- a/lti_consumer/outcomes.py +++ b/lti_consumer/outcomes.py @@ -6,16 +6,14 @@ """ 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 @@ def handle_request(self, request): 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 9128b6f8..00000000 --- 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 ea376588..4c4a9131 100644 --- a/lti_consumer/tests/unit/test_lti_consumer.py +++ b/lti_consumer/tests/unit/test_lti_consumer.py @@ -4,16 +4,15 @@ 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 @@ def test_context_id(self): """ 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 @@ def test_prefixed_custom_parameters(self): """ 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 @@ def test_lti_1p3_fields_appear_when_enabled(self, lti_1p3_enabled_mock): 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 @@ def setUp(self): 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 @@ def test_accept_grades_past_due_false_and_is_past_due_true(self, mock_is_past_du 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 @@ def test_parse_suffix_raises_error(self, mock_parse_suffix): 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 @@ def test_bad_user_id(self, mock_parse_suffix): 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 @@ def test_bad_request_method(self, mock_parse_suffix): 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 @@ def test_access_token_malformed(self, *args, **kwargs): 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 @@ def test_access_token_invalid_grant(self, *args, **kwargs): 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 @@ def test_access_token_invalid_client(self, *args, **kwargs): 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 @@ def test_access_token(self, *args, **kwargs): """ 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 fdf57de2..0ad41da9 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 5aa07acf..c2bccb50 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 15734d92..adf045c4 100644 --- a/setup.py +++ b/setup.py @@ -49,7 +49,7 @@ def is_requirement(line): 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',