-
- LTIBlock: {% trans "count is now" %} {{ count }} {% trans "click me to increment." %}
-
+
+
+
+ {% if has_score and weight %}
+
+ {% if module_score is not None %}
+ {# Translators: "points" is the student's achieved score on this LTI unit, and "total_points" is the maximum number of points achievable. #}
+ {% trans "points" as points_trans %}
+ {% trans "{points} / {total_points} points" as score_trans %}
+ ({{ module_score }} / {{ weight }} {{ points_trans }})
+ {% else %}
+ {# Translators: "total_points" is the maximum number of points achievable on this LTI unit #}
+ {% trans "total_points" as total_points_trans %}
+ ({{ weight }} {{ total_points_trans }} possible)
+ {% endif %}
+
+ {% endif %}
+
+
+
+ {% if launch_url and launch_url != 'http://www.example.com' and not hide_launch %}
+ {% if open_in_a_new_page %}
+
+ {% else %}
+ {# The result of the form submit will be rendered here. #}
+
+ {% endif %}
+ {% elif not hide_launch %}
+
+ {{ _('Please provide launch_url. Click "Edit", and fill in the required fields.') }}
+
+ {% endif %}
+
+ {% if has_score and comment %}
+
${_("Feedback on your work from the grader:")}
+
+ {# sanitized with nh3 in view #}
+ {{ comment|safe }}
+
+ {% endif %}
+
+
diff --git a/xblocks_contrib/lti/templates/lti_form.html b/xblocks_contrib/lti/templates/lti_form.html
new file mode 100644
index 00000000..5b7d12aa
--- /dev/null
+++ b/xblocks_contrib/lti/templates/lti_form.html
@@ -0,0 +1,38 @@
+{% load i18n %}
+
+
+
+
+
+
LTI
+
+
+ {% comment %}
+ This form will be hidden.
+ LTI block JavaScript will trigger a "submit" on the form, and the
+ result will be rendered instead.
+ {% endcomment %}
+
+
+
+
diff --git a/xblocks_contrib/lti/tests/__init__.py b/xblocks_contrib/lti/tests/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/xblocks_contrib/lti/tests/helpers.py b/xblocks_contrib/lti/tests/helpers.py
new file mode 100644
index 00000000..23c71618
--- /dev/null
+++ b/xblocks_contrib/lti/tests/helpers.py
@@ -0,0 +1,167 @@
+"""
+Utility methods for unit tests.
+"""
+
+import datetime
+import re
+from unittest.mock import Mock
+
+from xblock.fields import JSONField
+from xblock.reference.user_service import UserService, XBlockUser
+from xblock.runtime import Runtime
+
+TIMEDELTA_REGEX = re.compile(
+ r'^'
+ r'((?P
\d+?) day(?:s?))?(\s)?'
+ r'((?P\d+?) hour(?:s?))?(\s)?'
+ r'((?P\d+?) minute(?:s)?)?(\s)?'
+ r'((?P\d+?) second(?:s)?)?'
+ r'$'
+)
+
+
+class Timedelta(JSONField): # lint-amnesty, pylint: disable=missing-class-docstring
+ # Timedeltas are immutable, see http://docs.python.org/2/library/datetime.html#available-types
+ MUTABLE = False
+
+ def from_json(self, time_str): # lint-amnesty, pylint: disable=arguments-renamed, inconsistent-return-statements
+ """
+ time_str: A string with the following components:
+ day[s] (optional)
+ hour[s] (optional)
+ minute[s] (optional)
+ second[s] (optional)
+
+ Returns a datetime.timedelta parsed from the string
+ """
+ if time_str is None:
+ return None
+
+ if isinstance(time_str, datetime.timedelta):
+ return time_str
+
+ parts = TIMEDELTA_REGEX.match(time_str)
+ if not parts:
+ return
+ parts = parts.groupdict()
+ time_params = {}
+ for (name, param) in parts.items():
+ if param:
+ time_params[name] = int(param)
+ return datetime.timedelta(**time_params)
+
+ def to_json(self, value):
+ if value is None:
+ return None
+
+ values = []
+ for attr in ('days', 'hours', 'minutes', 'seconds'):
+ cur_value = getattr(value, attr, 0)
+ if cur_value > 0:
+ values.append("%d %s" % (cur_value, attr))
+ return ' '.join(values)
+
+ def enforce_type(self, value):
+ """
+ Ensure that when set explicitly the Field is set to a timedelta
+ """
+ if isinstance(value, datetime.timedelta) or value is None:
+ return value
+
+ return self.from_json(value)
+
+
+class StubUserService(UserService):
+ """
+ Stub UserService for testing the sequence block.
+ """
+
+ def __init__(self, # pylint: disable=too-many-positional-arguments
+ user=None,
+ user_is_staff=False,
+ user_role=None,
+ anonymous_user_id=None,
+ deprecated_anonymous_user_id=None,
+ request_country_code=None,
+ **kwargs):
+ self.user = user
+ self.user_is_staff = user_is_staff
+ self.user_role = user_role
+ self.anonymous_user_id = anonymous_user_id
+ self.deprecated_anonymous_user_id = deprecated_anonymous_user_id
+ self.request_country_code = request_country_code
+ self._django_user = user
+ super().__init__(**kwargs)
+
+ def get_current_user(self):
+ """
+ Implements abstract method for getting the current user.
+ """
+ user = XBlockUser()
+ if self.user and self.user.is_authenticated:
+ user.opt_attrs['edx-platform.anonymous_user_id'] = self.anonymous_user_id
+ user.opt_attrs['edx-platform.deprecated_anonymous_user_id'] = self.deprecated_anonymous_user_id
+ user.opt_attrs['edx-platform.request_country_code'] = self.request_country_code
+ user.opt_attrs['edx-platform.user_is_staff'] = self.user_is_staff
+ user.opt_attrs['edx-platform.user_id'] = self.user.id
+ user.opt_attrs['edx-platform.user_role'] = self.user_role
+ user.opt_attrs['edx-platform.username'] = self.user.username
+ else:
+ user.opt_attrs['edx-platform.username'] = 'anonymous'
+ user.opt_attrs['edx-platform.request_country_code'] = self.request_country_code
+ user.opt_attrs['edx-platform.is_authenticated'] = False
+ return user
+
+ def get_user_by_anonymous_id(self, uid=None): # pylint: disable=unused-argument
+ """
+ Return the original user passed into the service.
+ """
+ return self.user
+
+
+class MockRuntime(Runtime): # pylint: disable=abstract-method
+ """A mock implementation of the Runtime class for testing purposes."""
+
+ def __init__(self, anonymous_student_id, services=None):
+ # id_reader and id_generator are required by Runtime.
+ super().__init__(id_reader=lambda: None, id_generator=lambda: None, services=services)
+ self.anonymous_student_id = anonymous_student_id
+
+ def handler_url(
+ self, block, handler_name, suffix="", query="", thirdparty=False
+ ): # pylint: disable=too-many-positional-arguments
+ return f"/mock_url/{handler_name}"
+
+ def local_resource_url(self, block, resource): # pylint: disable=arguments-renamed
+ return f"/mock_resource_url/{resource}"
+
+ def resource_url(self, resource):
+ return f"/mock_resource/{resource}"
+
+ def publish(self, block, event_type, event_data):
+ pass
+
+
+def get_test_system(
+ user=None,
+ user_is_staff=False,
+):
+ """Construct a minimal test system for the LTIBlockTest."""
+
+ if not user:
+ user = Mock(name='get_test_system.user', is_staff=False)
+ user_service = StubUserService(
+ user=user,
+ anonymous_user_id='student',
+ deprecated_anonymous_user_id='student',
+ user_is_staff=user_is_staff,
+ user_role='student',
+ )
+ runtime = MockRuntime(
+ anonymous_student_id="student",
+ services={
+ "user": user_service,
+ }
+ )
+
+ return runtime
diff --git a/xblocks_contrib/lti/tests/test_lti.py b/xblocks_contrib/lti/tests/test_lti.py
deleted file mode 100644
index f4a4a53d..00000000
--- a/xblocks_contrib/lti/tests/test_lti.py
+++ /dev/null
@@ -1,26 +0,0 @@
-"""
-Tests for LTIBlock
-"""
-
-from django.test import TestCase
-from xblock.fields import ScopeIds
-from xblock.test.toy_runtime import ToyRuntime
-
-from xblocks_contrib import LTIBlock
-
-
-class TestLTIBlock(TestCase):
- """Tests for LTIBlock"""
-
- def test_my_student_view(self):
- """Test the basic view loads."""
- scope_ids = ScopeIds("1", "2", "3", "4")
- block = LTIBlock(ToyRuntime(), scope_ids=scope_ids)
- frag = block.student_view()
- as_dict = frag.to_dict()
- content = as_dict["content"]
- self.assertIn(
- "LTIBlock: count is now",
- content,
- "XBlock did not render correct student view",
- )
diff --git a/xblocks_contrib/lti/tests/test_lti20_unit.py b/xblocks_contrib/lti/tests/test_lti20_unit.py
new file mode 100644
index 00000000..efad5cc9
--- /dev/null
+++ b/xblocks_contrib/lti/tests/test_lti20_unit.py
@@ -0,0 +1,428 @@
+"""Tests for LTI Xmodule LTIv2.0 functional logic."""
+
+import datetime
+import textwrap
+from unittest.mock import Mock
+
+from django.conf import settings
+from django.test import TestCase, override_settings
+from pytz import UTC
+from xblock.field_data import DictFieldData
+
+from xblocks_contrib.lti.lti import LTIBlock
+from xblocks_contrib.lti.lti_2_util import LTIError
+
+from .helpers import StubUserService, get_test_system
+
+
+@override_settings(LMS_BASE="edx.org")
+class LTI20RESTResultServiceTest(TestCase):
+ """Logic tests for LTI block. LTI2.0 REST ResultService"""
+
+ USER_STANDIN = Mock()
+ USER_STANDIN.id = 999
+
+ def setUp(self):
+ super().setUp()
+ self.runtime = get_test_system(user=self.USER_STANDIN)
+ self.environ = {"wsgi.url_scheme": "http", "REQUEST_METHOD": "POST"}
+ self.runtime.publish = Mock()
+ self.runtime._services["rebind_user"] = Mock() # pylint: disable=protected-access
+
+ self.xblock = LTIBlock(self.runtime, DictFieldData({}), Mock())
+ self.lti_id = self.xblock.lti_id
+
+ self.unquoted_resource_link_id = (
+ "{}-i4x-2-3-lti-31de800015cf4afb973356dbe81496df".format(settings.LMS_BASE)
+ )
+
+ self.xblock.due = None
+ self.xblock.graceperiod = None
+
+ def test_sanitize_get_context(self):
+ """Tests that the get_context function does basic sanitization"""
+ # get_context, unfortunately, requires a lot of mocking machinery
+ mocked_course = Mock(
+ name="mocked_course", lti_passports=["lti_id:test_client:test_secret"]
+ )
+ modulestore = Mock(name="modulestore")
+ modulestore.get_course.return_value = mocked_course
+ self.xblock.runtime.modulestore = modulestore
+ self.xblock.lti_id = "lti_id"
+
+ test_cases = ( # (before sanitize, after sanitize)
+ ("plaintext", "plaintext"),
+ ("a ", "a "), # drops scripts
+ ("bold 包", "bold 包"), # unicode, and tags pass through
+ )
+ for case in test_cases:
+ self.xblock.score_comment = case[0]
+ assert case[1] == self.xblock.get_context()["comment"]
+
+ def test_lti20_rest_bad_contenttype(self):
+ """
+ Input with bad content type
+ """
+ with self.assertRaisesRegex(LTIError, "Content-Type must be"):
+ request = Mock(headers={"Content-Type": "Non-existent"})
+ self.xblock.verify_lti_2_0_result_rest_headers(request)
+
+ def test_lti20_rest_failed_oauth_body_verify(self):
+ """
+ Input with bad oauth body hash verification
+ """
+ err_msg = "OAuth body verification failed"
+ self.xblock.verify_oauth_body_sign = Mock(side_effect=LTIError(err_msg))
+ with self.assertRaisesRegex(LTIError, err_msg):
+ request = Mock(
+ headers={"Content-Type": "application/vnd.ims.lis.v2.result+json"}
+ )
+ self.xblock.verify_lti_2_0_result_rest_headers(request)
+
+ def test_lti20_rest_good_headers(self):
+ """
+ Input with good oauth body hash verification
+ """
+ self.xblock.verify_oauth_body_sign = Mock(return_value=True)
+
+ request = Mock(
+ headers={"Content-Type": "application/vnd.ims.lis.v2.result+json"}
+ )
+ self.xblock.verify_lti_2_0_result_rest_headers(request)
+ # We just want the above call to complete without exceptions, and to have called verify_oauth_body_sign
+ assert self.xblock.verify_oauth_body_sign.called
+
+ BAD_DISPATCH_INPUTS = [
+ None,
+ "",
+ "abcd"
+ "notuser/abcd"
+ "user/"
+ "user//"
+ "user/gbere/"
+ "user/gbere/xsdf"
+ "user/ಠ益ಠ", # not alphanumeric
+ ]
+
+ def test_lti20_rest_bad_dispatch(self):
+ """
+ Test the error cases for the "dispatch" argument to the LTI 2.0 handler. Anything that doesn't
+ fit the form user/
+ """
+ for einput in self.BAD_DISPATCH_INPUTS:
+ with self.assertRaisesRegex(LTIError, "No valid user id found in endpoint URL"):
+ self.xblock.parse_lti_2_0_handler_suffix(einput)
+
+ GOOD_DISPATCH_INPUTS = [
+ ("user/abcd3", "abcd3"),
+ ("user/Äbcdè2", "Äbcdè2"), # unicode, just to make sure
+ ]
+
+ def test_lti20_rest_good_dispatch(self):
+ """
+ Test the good cases for the "dispatch" argument to the LTI 2.0 handler. Anything that does
+ fit the form user/
+ """
+ for ginput, expected in self.GOOD_DISPATCH_INPUTS:
+ assert self.xblock.parse_lti_2_0_handler_suffix(ginput) == expected
+
+ BAD_JSON_INPUTS = [
+ # (bad inputs, error message expected)
+ ([
+ "kk", # ValueError
+ "{{}", # ValueError
+ "{}}", # ValueError
+ 3, # TypeError
+ {}, # TypeError
+ ], "Supplied JSON string in request body could not be decoded"),
+ ([
+ "3", # valid json, not array or object
+ "[]", # valid json, array too small
+ "[3, {}]", # valid json, 1st element not an object
+ ], "Supplied JSON string is a list that does not contain an object as the first element"),
+ ([
+ '{"@type": "NOTResult"}', # @type key must have value 'Result'
+ ], "JSON object does not contain correct @type attribute"),
+ ([
+ # @context missing
+ '{"@type": "Result", "resultScore": 0.1}',
+ ], "JSON object does not contain required key"),
+ ([
+ '''
+ {"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "resultScore": 100}''' # score out of range
+ ], "score value outside the permitted range of 0-1."),
+ ([
+ '''
+ {"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "resultScore": "1b"}''', # score ValueError
+ '''
+ {"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "resultScore": {}}''', # score TypeError
+ ], "Could not convert resultScore to float"),
+ ]
+
+ def test_lti20_bad_json(self):
+ """
+ Test that bad json_str to parse_lti_2_0_result_json inputs raise appropriate LTI Error
+ """
+ for error_inputs, error_message in self.BAD_JSON_INPUTS:
+ for einput in error_inputs:
+ with self.assertRaisesRegex(LTIError, error_message):
+ self.xblock.parse_lti_2_0_result_json(einput)
+
+ GOOD_JSON_INPUTS = [
+ ('''
+ {"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "resultScore": 0.1}''', ""), # no comment means we expect ""
+ ('''
+ [{"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "@id": "anon_id:abcdef0123456789",
+ "resultScore": 0.1}]''', ""), # OK to have array of objects -- just take the first. @id is okay too
+ ('''
+ {"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "resultScore": 0.1,
+ "comment": "ಠ益ಠ"}''', "ಠ益ಠ"), # unicode comment
+ ]
+
+ def test_lti20_good_json(self):
+ """
+ Test the parsing of good comments
+ """
+ for json_str, expected_comment in self.GOOD_JSON_INPUTS:
+ score, comment = self.xblock.parse_lti_2_0_result_json(json_str)
+ assert score == 0.1
+ assert comment == expected_comment
+
+ GOOD_JSON_PUT = textwrap.dedent("""
+ {"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "@id": "anon_id:abcdef0123456789",
+ "resultScore": 0.1,
+ "comment": "ಠ益ಠ"}
+ """).encode('utf-8')
+
+ GOOD_JSON_PUT_LIKE_DELETE = textwrap.dedent("""
+ {"@type": "Result",
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "@id": "anon_id:abcdef0123456789",
+ "comment": "ಠ益ಠ"}
+ """).encode('utf-8')
+
+ def get_signed_lti20_mock_request(self, body, method="PUT"):
+ """
+ Example of signed from LTI 2.0 Provider. Signatures and hashes are example only and won't verify
+ """
+ mock_request = Mock()
+ mock_request.headers = {
+ 'Content-Type': 'application/vnd.ims.lis.v2.result+json',
+ 'Authorization': (
+ 'OAuth oauth_nonce="135685044251684026041377608307", '
+ 'oauth_timestamp="1234567890", oauth_version="1.0", '
+ 'oauth_signature_method="HMAC-SHA1", '
+ 'oauth_consumer_key="test_client_key", '
+ 'oauth_signature="my_signature%3D", '
+ 'oauth_body_hash="gz+PeJZuF2//n9hNUnDj2v5kN70="'
+ ),
+ }
+ mock_request.url = "http://testurl"
+ mock_request.http_method = method
+ mock_request.method = method
+ mock_request.body = body
+ return mock_request
+
+ def setup_system_xblock_mocks_for_lti20_request_test(self):
+ """
+ Helper fn to set up mocking for lti 2.0 request test
+ """
+ self.xblock.max_score = Mock(return_value=1.0)
+ self.xblock.get_client_key_secret = Mock(
+ return_value=("test_client_key", "test_client_secret")
+ )
+ self.xblock.verify_oauth_body_sign = Mock()
+
+ def test_lti20_put_like_delete_success(self):
+ """
+ The happy path for LTI 2.0 PUT that acts like a delete
+ """
+ self.setup_system_xblock_mocks_for_lti20_request_test()
+ SCORE = 0.55 # pylint: disable=invalid-name
+ COMMENT = "ಠ益ಠ" # pylint: disable=invalid-name
+ self.xblock.module_score = SCORE
+ self.xblock.score_comment = COMMENT
+ mock_request = self.get_signed_lti20_mock_request(
+ self.GOOD_JSON_PUT_LIKE_DELETE
+ )
+ # Now call the handler
+ response = self.xblock.lti_2_0_result_rest_handler(mock_request, "user/abcd")
+ # Now assert there's no score
+ assert response.status_code == 200
+ assert self.xblock.module_score is None
+ assert self.xblock.score_comment == ""
+ (_, evt_type, called_grade_obj), _ = (
+ self.runtime.publish.call_args
+ )
+ assert called_grade_obj == {
+ "user_id": self.USER_STANDIN.id,
+ "value": None,
+ "max_value": None,
+ "score_deleted": True,
+ }
+ assert evt_type == "grade"
+
+ def test_lti20_delete_success(self):
+ """
+ The happy path for LTI 2.0 DELETE
+ """
+ self.setup_system_xblock_mocks_for_lti20_request_test()
+ SCORE = 0.55 # pylint: disable=invalid-name
+ COMMENT = "ಠ益ಠ" # pylint: disable=invalid-name
+ self.xblock.module_score = SCORE
+ self.xblock.score_comment = COMMENT
+ mock_request = self.get_signed_lti20_mock_request(b"", method="DELETE")
+ # Now call the handler
+ response = self.xblock.lti_2_0_result_rest_handler(mock_request, "user/abcd")
+ # Now assert there's no score
+ assert response.status_code == 200
+ assert self.xblock.module_score is None
+ assert self.xblock.score_comment == ""
+ (_, evt_type, called_grade_obj), _ = (
+ self.runtime.publish.call_args
+ )
+ assert called_grade_obj == {
+ "user_id": self.USER_STANDIN.id,
+ "value": None,
+ "max_value": None,
+ "score_deleted": True,
+ }
+ assert evt_type == "grade"
+
+ def test_lti20_put_set_score_success(self):
+ """
+ The happy path for LTI 2.0 PUT that sets a score
+ """
+ self.setup_system_xblock_mocks_for_lti20_request_test()
+ mock_request = self.get_signed_lti20_mock_request(self.GOOD_JSON_PUT)
+ # Now call the handler
+ response = self.xblock.lti_2_0_result_rest_handler(mock_request, "user/abcd")
+ # Now assert
+ assert response.status_code == 200
+ assert self.xblock.module_score == 0.1
+ assert self.xblock.score_comment == "ಠ益ಠ"
+ (_, evt_type, called_grade_obj), _ = (
+ self.runtime.publish.call_args
+ )
+ assert evt_type == "grade"
+ assert called_grade_obj == {
+ "user_id": self.USER_STANDIN.id,
+ "value": 0.1,
+ "max_value": 1.0,
+ "score_deleted": False,
+ }
+
+ def test_lti20_get_no_score_success(self):
+ """
+ The happy path for LTI 2.0 GET when there's no score
+ """
+ self.setup_system_xblock_mocks_for_lti20_request_test()
+ mock_request = self.get_signed_lti20_mock_request(b"", method="GET")
+ # Now call the handler
+ response = self.xblock.lti_2_0_result_rest_handler(mock_request, "user/abcd")
+ # Now assert
+ assert response.status_code == 200
+ assert response.json == {
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "@type": "Result",
+ }
+
+ def test_lti20_get_with_score_success(self):
+ """
+ The happy path for LTI 2.0 GET when there is a score
+ """
+ self.setup_system_xblock_mocks_for_lti20_request_test()
+ SCORE = 0.55 # pylint: disable=invalid-name
+ COMMENT = "ಠ益ಠ" # pylint: disable=invalid-name
+ self.xblock.module_score = SCORE
+ self.xblock.score_comment = COMMENT
+ mock_request = self.get_signed_lti20_mock_request(b"", method="GET")
+ # Now call the handler
+ response = self.xblock.lti_2_0_result_rest_handler(mock_request, "user/abcd")
+ # Now assert
+ assert response.status_code == 200
+ assert response.json == {
+ "@context": "http://purl.imsglobal.org/ctx/lis/v2/Result",
+ "@type": "Result",
+ "resultScore": SCORE,
+ "comment": COMMENT,
+ }
+
+ UNSUPPORTED_HTTP_METHODS = ["OPTIONS", "HEAD", "POST", "TRACE", "CONNECT"]
+
+ def test_lti20_unsupported_method_error(self):
+ """
+ Test we get a 404 when we don't GET or PUT
+ """
+ self.setup_system_xblock_mocks_for_lti20_request_test()
+ mock_request = self.get_signed_lti20_mock_request(self.GOOD_JSON_PUT)
+ for bad_method in self.UNSUPPORTED_HTTP_METHODS:
+ mock_request.method = bad_method
+ response = self.xblock.lti_2_0_result_rest_handler(
+ mock_request, "user/abcd"
+ )
+ assert response.status_code == 404
+
+ def test_lti20_request_handler_bad_headers(self):
+ """
+ Test that we get a 401 when header verification fails
+ """
+ self.setup_system_xblock_mocks_for_lti20_request_test()
+ self.xblock.verify_lti_2_0_result_rest_headers = Mock(side_effect=LTIError())
+ mock_request = self.get_signed_lti20_mock_request(self.GOOD_JSON_PUT)
+ response = self.xblock.lti_2_0_result_rest_handler(mock_request, "user/abcd")
+ assert response.status_code == 401
+
+ def test_lti20_request_handler_bad_dispatch_user(self):
+ """
+ Test that we get a 404 when there's no (or badly formatted) user specified in the url
+ """
+ self.setup_system_xblock_mocks_for_lti20_request_test()
+ mock_request = self.get_signed_lti20_mock_request(self.GOOD_JSON_PUT)
+ response = self.xblock.lti_2_0_result_rest_handler(mock_request, None)
+ assert response.status_code == 404
+
+ def test_lti20_request_handler_bad_json(self):
+ """
+ Test that we get a 404 when json verification fails
+ """
+ self.setup_system_xblock_mocks_for_lti20_request_test()
+ self.xblock.parse_lti_2_0_result_json = Mock(side_effect=LTIError())
+ mock_request = self.get_signed_lti20_mock_request(self.GOOD_JSON_PUT)
+ response = self.xblock.lti_2_0_result_rest_handler(mock_request, "user/abcd")
+ assert response.status_code == 404
+
+ def test_lti20_request_handler_bad_user(self):
+ """
+ Test that we get a 404 when the supplied user does not exist
+ """
+ self.setup_system_xblock_mocks_for_lti20_request_test()
+ self.runtime._services["user"] = StubUserService(user=None) # pylint: disable=protected-access
+ mock_request = self.get_signed_lti20_mock_request(self.GOOD_JSON_PUT)
+ response = self.xblock.lti_2_0_result_rest_handler(mock_request, "user/abcd")
+ assert response.status_code == 404
+
+ def test_lti20_request_handler_grade_past_due(self):
+ """
+ Test that we get a 404 when accept_grades_past_due is False and it is past due
+ """
+ self.setup_system_xblock_mocks_for_lti20_request_test()
+ self.xblock.due = datetime.datetime.now(UTC)
+ self.xblock.accept_grades_past_due = False
+ mock_request = self.get_signed_lti20_mock_request(self.GOOD_JSON_PUT)
+ response = self.xblock.lti_2_0_result_rest_handler(mock_request, "user/abcd")
+ assert response.status_code == 404
diff --git a/xblocks_contrib/lti/tests/test_lti_unit.py b/xblocks_contrib/lti/tests/test_lti_unit.py
new file mode 100644
index 00000000..cbefdda9
--- /dev/null
+++ b/xblocks_contrib/lti/tests/test_lti_unit.py
@@ -0,0 +1,579 @@
+"""Test for LTI Xmodule functional logic."""
+
+import datetime
+import textwrap
+from copy import copy
+from unittest.mock import Mock, PropertyMock, patch
+from urllib import parse
+
+import pytest
+from django.conf import settings
+from django.test import TestCase, override_settings
+from lxml import etree
+from opaque_keys.edx.keys import CourseKey
+from opaque_keys.edx.locator import BlockUsageLocator
+from pytz import UTC
+from webob.request import Request
+from xblock.field_data import DictFieldData
+from xblock.fields import ScopeIds
+
+from xblocks_contrib.lti.lti import LTIBlock
+from xblocks_contrib.lti.lti_2_util import LTIError
+
+from .helpers import StubUserService, Timedelta, get_test_system
+
+ATTR_KEY_ANONYMOUS_USER_ID = "edx-platform.anonymous_user_id"
+
+
+@override_settings(LMS_BASE="edx.org")
+class LTIBlockTest(TestCase):
+ """Logic tests for LTI block."""
+
+ def setUp(self):
+ super().setUp()
+ self.environ = {"wsgi.url_scheme": "http", "REQUEST_METHOD": "POST"}
+ self.request_body_xml_template = textwrap.dedent(
+ """
+
+
+
+
+ V1.0
+ {messageIdentifier}
+
+
+
+ <{action}>
+
+
+ {sourcedId}
+
+
+
+ en-us
+ {grade}
+
+
+
+
+
+
+ """)
+ self.course_id = CourseKey.from_string('org/course/run')
+ self.runtime = get_test_system()
+ self.runtime.publish = Mock()
+ self.runtime._services['rebind_user'] = Mock() # pylint: disable=protected-access
+
+ self.xblock = LTIBlock(
+ self.runtime,
+ DictFieldData({}),
+ ScopeIds(
+ None, None, None, BlockUsageLocator(self.course_id, "lti", "name")
+ ),
+ )
+ current_user = self.runtime.service(self.xblock, "user").get_current_user()
+ self.user_id = current_user.opt_attrs.get(ATTR_KEY_ANONYMOUS_USER_ID)
+ self.lti_id = self.xblock.lti_id
+
+ self.unquoted_resource_link_id = (
+ "{}-i4x-2-3-lti-31de800015cf4afb973356dbe81496df".format(settings.LMS_BASE)
+ )
+
+ sourced_id = ":".join(
+ parse.quote(i)
+ for i in (self.lti_id, self.unquoted_resource_link_id, self.user_id)
+ )
+
+ self.defaults = {
+ "namespace": "http://www.imsglobal.org/services/ltiv1p1/xsd/imsoms_v1p0",
+ "sourcedId": sourced_id,
+ "action": "replaceResultRequest",
+ "grade": 0.5,
+ "messageIdentifier": "528243ba5241b",
+ }
+
+ self.xblock.due = None
+ self.xblock.graceperiod = None
+
+ def get_request_body(self, params=None):
+ """Fetches the body of a request specified by params"""
+ if params is None:
+ params = {}
+ data = copy(self.defaults)
+
+ data.update(params)
+ return self.request_body_xml_template.format(**data).encode("utf-8")
+
+ def get_response_values(self, response):
+ """Gets the values from the given response"""
+ parser = etree.XMLParser(ns_clean=True, recover=True, encoding="utf-8")
+ root = etree.fromstring(response.body.strip(), parser=parser)
+ lti_spec_namespace = "http://www.imsglobal.org/services/ltiv1p1/xsd/imsoms_v1p0"
+ namespaces = {"def": lti_spec_namespace}
+
+ code_major = root.xpath("//def:imsx_codeMajor", namespaces=namespaces)[0].text
+ description = root.xpath("//def:imsx_description", namespaces=namespaces)[
+ 0
+ ].text
+ message_identifier = root.xpath(
+ "//def:imsx_messageIdentifier", namespaces=namespaces
+ )[0].text
+ imsx_pox_body = root.xpath("//def:imsx_POXBody", namespaces=namespaces)[0]
+
+ try:
+ action = imsx_pox_body.getchildren()[0].tag.replace(
+ "{" + lti_spec_namespace + "}", ""
+ )
+ except Exception: # pylint: disable=broad-except
+ action = None
+
+ return {
+ "code_major": code_major,
+ "description": description,
+ "messageIdentifier": message_identifier,
+ "action": action,
+ }
+
+ @patch(
+ "xblocks_contrib.lti.lti.LTIBlock.get_client_key_secret",
+ return_value=("test_client_key", "test_client_secret"),
+ )
+ def test_authorization_header_not_present(self, _get_key_secret):
+ """
+ Request has no Authorization header.
+
+ This is an unknown service request, i.e., it is not a part of the original service specification.
+ """
+ request = Request(self.environ)
+ request.body = self.get_request_body()
+ response = self.xblock.grade_handler(request, "")
+ real_response = self.get_response_values(response)
+ expected_response = {
+ "action": None,
+ "code_major": "failure",
+ "description": "OAuth verification error: Malformed authorization header",
+ "messageIdentifier": self.defaults["messageIdentifier"],
+ }
+
+ assert response.status_code == 200
+ self.assertDictEqual(expected_response, real_response)
+
+ @patch(
+ "xblocks_contrib.lti.lti.LTIBlock.get_client_key_secret",
+ return_value=("test_client_key", "test_client_secret"),
+ )
+ def test_authorization_header_empty(self, _get_key_secret):
+ """
+ Request Authorization header has no value.
+
+ This is an unknown service request, i.e., it is not a part of the original service specification.
+ """
+ request = Request(self.environ)
+ request.authorization = "bad authorization header"
+ request.body = self.get_request_body()
+ response = self.xblock.grade_handler(request, "")
+ real_response = self.get_response_values(response)
+ expected_response = {
+ "action": None,
+ "code_major": "failure",
+ "description": "OAuth verification error: Malformed authorization header",
+ "messageIdentifier": self.defaults["messageIdentifier"],
+ }
+ assert response.status_code == 200
+ self.assertDictEqual(expected_response, real_response)
+
+ def test_real_user_is_none(self):
+ """
+ If we have no real user, we should send back failure response.
+ """
+ self.runtime._services["user"] = StubUserService(user=None) # pylint: disable=protected-access
+ self.xblock.verify_oauth_body_sign = Mock()
+ self.xblock.has_score = True
+ request = Request(self.environ)
+ request.body = self.get_request_body()
+ response = self.xblock.grade_handler(request, "")
+ real_response = self.get_response_values(response)
+ expected_response = {
+ "action": None,
+ "code_major": "failure",
+ "description": "User not found.",
+ "messageIdentifier": self.defaults["messageIdentifier"],
+ }
+ assert response.status_code == 200
+ self.assertDictEqual(expected_response, real_response)
+
+ def test_grade_past_due(self):
+ """
+ Should fail if we do not accept past due grades, and it is past due.
+ """
+ self.xblock.accept_grades_past_due = False
+ self.xblock.due = datetime.datetime.now(UTC)
+ self.xblock.graceperiod = Timedelta().from_json("0 seconds")
+ request = Request(self.environ)
+ request.body = self.get_request_body()
+ response = self.xblock.grade_handler(request, "")
+ real_response = self.get_response_values(response)
+ expected_response = {
+ "action": None,
+ "code_major": "failure",
+ "description": "Grade is past due",
+ "messageIdentifier": "unknown",
+ }
+ assert response.status_code == 200
+ assert expected_response == real_response
+
+ def test_grade_not_in_range(self):
+ """
+ Grade returned from Tool Provider is outside the range 0.0-1.0.
+ """
+ self.xblock.verify_oauth_body_sign = Mock()
+ request = Request(self.environ)
+ request.body = self.get_request_body(params={"grade": "10"})
+ response = self.xblock.grade_handler(request, "")
+ real_response = self.get_response_values(response)
+ expected_response = {
+ "action": None,
+ "code_major": "failure",
+ "description": "Request body XML parsing error: score value outside the permitted range of 0-1.",
+ "messageIdentifier": "unknown",
+ }
+ assert response.status_code == 200
+ self.assertDictEqual(expected_response, real_response)
+
+ def test_bad_grade_decimal(self):
+ """
+ Grade returned from Tool Provider doesn't use a period as the decimal point.
+ """
+ self.xblock.verify_oauth_body_sign = Mock()
+ request = Request(self.environ)
+ request.body = self.get_request_body(params={"grade": "0,5"})
+ response = self.xblock.grade_handler(request, "")
+ real_response = self.get_response_values(response)
+ msg = "could not convert string to float: '0,5'"
+ expected_response = {
+ "action": None,
+ "code_major": "failure",
+ "description": f"Request body XML parsing error: {msg}",
+ "messageIdentifier": "unknown",
+ }
+ assert response.status_code == 200
+ self.assertDictEqual(expected_response, real_response)
+
+ def test_unsupported_action(self):
+ """
+ Action returned from Tool Provider isn't supported.
+ `replaceResultRequest` is supported only.
+ """
+ self.xblock.verify_oauth_body_sign = Mock()
+ request = Request(self.environ)
+ request.body = self.get_request_body({"action": "wrongAction"})
+ response = self.xblock.grade_handler(request, "")
+ real_response = self.get_response_values(response)
+ expected_response = {
+ "action": None,
+ "code_major": "unsupported",
+ "description": "Target does not support the requested operation.",
+ "messageIdentifier": self.defaults["messageIdentifier"],
+ }
+ assert response.status_code == 200
+ self.assertDictEqual(expected_response, real_response)
+
+ def test_good_request(self):
+ """
+ Response from Tool Provider is correct.
+ """
+ self.xblock.verify_oauth_body_sign = Mock()
+ self.xblock.has_score = True
+ request = Request(self.environ)
+ request.body = self.get_request_body()
+ response = self.xblock.grade_handler(request, "")
+ description_expected = "Score for {sourcedId} is now {score}".format(
+ sourcedId=self.defaults["sourcedId"],
+ score=self.defaults["grade"],
+ )
+ real_response = self.get_response_values(response)
+ expected_response = {
+ "action": "replaceResultResponse",
+ "code_major": "success",
+ "description": description_expected,
+ "messageIdentifier": self.defaults["messageIdentifier"],
+ }
+
+ assert response.status_code == 200
+ self.assertDictEqual(expected_response, real_response)
+ assert self.xblock.module_score == float(self.defaults["grade"])
+
+ def test_user_id(self):
+ expected_user_id = str(parse.quote(self.xblock.runtime.anonymous_student_id))
+ real_user_id = self.xblock.get_user_id()
+ assert real_user_id == expected_user_id
+
+ def test_outcome_service_url(self):
+ mock_url_prefix = "https://hostname/"
+ test_service_name = "test_service"
+
+ def mock_handler_url(
+ block, handler_name, **kwargs
+ ): # pylint: disable=unused-argument
+ """Mock function for returning fully-qualified handler urls"""
+ return mock_url_prefix + handler_name
+
+ self.xblock.runtime.handler_url = Mock(side_effect=mock_handler_url)
+ real_outcome_service_url = self.xblock.get_outcome_service_url(
+ service_name=test_service_name
+ )
+ assert real_outcome_service_url == (mock_url_prefix + test_service_name)
+
+ def test_resource_link_id(self):
+ with patch(
+ "xblocks_contrib.lti.lti.LTIBlock.location", new_callable=PropertyMock
+ ):
+ self.xblock.location.html_id = (
+ lambda: "i4x-2-3-lti-31de800015cf4afb973356dbe81496df"
+ )
+ expected_resource_link_id = str(parse.quote(self.unquoted_resource_link_id))
+ real_resource_link_id = self.xblock.get_resource_link_id()
+ assert real_resource_link_id == expected_resource_link_id
+
+ def test_lis_result_sourcedid(self):
+ expected_sourced_id = ":".join(
+ parse.quote(i)
+ for i in (
+ str(self.course_id),
+ self.xblock.get_resource_link_id(),
+ self.user_id,
+ )
+ )
+ real_lis_result_sourcedid = self.xblock.get_lis_result_sourcedid()
+ assert real_lis_result_sourcedid == expected_sourced_id
+
+ def test_client_key_secret(self):
+ """
+ LTI block gets client key and secret provided.
+ """
+ # this adds lti passports to system
+ mocked_course = Mock(lti_passports=["lti_id:test_client:test_secret"])
+ modulestore = Mock()
+ modulestore.get_course.return_value = mocked_course
+ runtime = Mock(modulestore=modulestore)
+ self.xblock.runtime = runtime
+ self.xblock.lti_id = "lti_id"
+ key, secret = self.xblock.get_client_key_secret()
+ expected = ("test_client", "test_secret")
+ assert expected == (key, secret)
+
+ def test_client_key_secret_not_provided(self):
+ """
+ LTI block attempts to get client key and secret provided in cms.
+
+ There are key and secret but not for specific LTI.
+ """
+
+ # this adds lti passports to system
+ mocked_course = Mock(lti_passports=["test_id:test_client:test_secret"])
+ modulestore = Mock()
+ modulestore.get_course.return_value = mocked_course
+ runtime = Mock(modulestore=modulestore)
+ self.xblock.runtime = runtime
+ # set another lti_id
+ self.xblock.lti_id = "another_lti_id"
+ key_secret = self.xblock.get_client_key_secret()
+ expected = ("", "")
+ assert expected == key_secret
+
+ def test_bad_client_key_secret(self):
+ """
+ LTI block attempts to get client key and secret provided in cms.
+
+ There are key and secret provided in wrong format.
+ """
+ # this adds lti passports to system
+ mocked_course = Mock(lti_passports=["test_id_test_client_test_secret"])
+ modulestore = Mock()
+ modulestore.get_course.return_value = mocked_course
+ runtime = Mock(modulestore=modulestore)
+ self.xblock.runtime = runtime
+ self.xblock.lti_id = "lti_id"
+ with pytest.raises(LTIError):
+ self.xblock.get_client_key_secret()
+
+ @patch(
+ "xblocks_contrib.lti.lti.signature.verify_hmac_sha1", Mock(return_value=True)
+ )
+ @patch(
+ "xblocks_contrib.lti.lti.LTIBlock.get_client_key_secret",
+ Mock(return_value=("test_client_key", "test_client_secret")),
+ )
+ def test_successful_verify_oauth_body_sign(self):
+ """
+ Test if OAuth signing was successful.
+ """
+ self.xblock.verify_oauth_body_sign(self.get_signed_grade_mock_request())
+
+ @patch(
+ "xblocks_contrib.lti.lti.LTIBlock.get_outcome_service_url",
+ Mock(return_value="https://testurl/"),
+ )
+ @patch(
+ "xblocks_contrib.lti.lti.LTIBlock.get_client_key_secret",
+ Mock(return_value=("__consumer_key__", "__lti_secret__")),
+ )
+ def test_failed_verify_oauth_body_sign_proxy_mangle_url(self):
+ """
+ Oauth signing verify fail.
+ """
+ request = self.get_signed_grade_mock_request_with_correct_signature()
+ self.xblock.verify_oauth_body_sign(request)
+ # we should verify against get_outcome_service_url not
+ # request url proxy and load balancer along the way may
+ # change url presented to the method
+ request.url = "http://testurl/"
+ self.xblock.verify_oauth_body_sign(request)
+
+ def get_signed_grade_mock_request_with_correct_signature(self):
+ """
+ Generate a proper LTI request object
+ """
+ mock_request = Mock()
+ mock_request.headers = {
+ "X-Requested-With": "XMLHttpRequest",
+ "Content-Type": "application/x-www-form-urlencoded",
+ "Authorization": (
+ 'OAuth realm="https://testurl/", oauth_body_hash="wwzA3s8gScKD1VpJ7jMt9b%2BMj9Q%3D",'
+ 'oauth_nonce="18821463", oauth_timestamp="1409321145", '
+ 'oauth_consumer_key="__consumer_key__", oauth_signature_method="HMAC-SHA1", '
+ 'oauth_version="1.0", oauth_signature="fHsE1hhIz76/msUoMR3Lyb7Aou4%3D"'
+ ),
+ }
+ mock_request.url = "https://testurl"
+ mock_request.http_method = "POST"
+ mock_request.method = mock_request.http_method
+
+ mock_request.body = (
+ b"\n"
+ b''
+ b"V1.0"
+ b"edX_fix"
+ b""
+ b"MITxLTI/MITxLTI/201x:localhost%3A8000-i4x-MITxLTI-MITxLTI-lti-3751833a214a4f66a0d18f63234207f2"
+ b":363979ef768ca171b50f9d1bfb322131"
+ b"en0.32"
+ b""
+ )
+
+ return mock_request
+
+ def test_wrong_xml_namespace(self):
+ """
+ Test wrong XML Namespace.
+
+ Tests that tool provider returned grade back with wrong XML Namespace.
+ """
+ with pytest.raises(IndexError):
+ mocked_request = self.get_signed_grade_mock_request(
+ namespace_lti_v1p1=False
+ )
+ self.xblock.parse_grade_xml_body(mocked_request.body)
+
+ def test_parse_grade_xml_body(self):
+ """
+ Test XML request body parsing.
+
+ Tests that xml body was parsed successfully.
+ """
+ mocked_request = self.get_signed_grade_mock_request()
+ message_identifier, sourced_id, grade, action = (
+ self.xblock.parse_grade_xml_body(mocked_request.body)
+ )
+ assert self.defaults["messageIdentifier"] == message_identifier
+ assert self.defaults["sourcedId"] == sourced_id
+ assert self.defaults["grade"] == grade
+ assert self.defaults["action"] == action
+
+ @patch(
+ "xblocks_contrib.lti.lti.signature.verify_hmac_sha1", Mock(return_value=False)
+ )
+ @patch(
+ "xblocks_contrib.lti.lti.LTIBlock.get_client_key_secret",
+ Mock(return_value=("test_client_key", "test_client_secret")),
+ )
+ def test_failed_verify_oauth_body_sign(self):
+ """
+ Oauth signing verify fail.
+ """
+ with pytest.raises(LTIError):
+ req = self.get_signed_grade_mock_request()
+ self.xblock.verify_oauth_body_sign(req)
+
+ def get_signed_grade_mock_request(self, namespace_lti_v1p1=True):
+ """
+ Example of signed request from LTI Provider.
+
+ When `namespace_v1p0` is set to True then the default namespase from
+ LTI 1.1 will be used. Otherwise fake namespace will be added to XML.
+ """
+ mock_request = Mock()
+ mock_request.headers = {
+ "X-Requested-With": "XMLHttpRequest",
+ "Content-Type": "application/x-www-form-urlencoded",
+ "Authorization": 'OAuth oauth_nonce="135685044251684026041377608307", \
+ oauth_timestamp="1234567890", oauth_version="1.0", \
+ oauth_signature_method="HMAC-SHA1", \
+ oauth_consumer_key="test_client_key", \
+ oauth_signature="my_signature%3D", \
+ oauth_body_hash="JEpIArlNCeV4ceXxric8gJQCnBw="',
+ }
+ mock_request.url = "http://testurl"
+ mock_request.http_method = "POST"
+
+ params = {}
+ if not namespace_lti_v1p1:
+ params = {"namespace": "http://www.fakenamespace.com/fake"}
+ mock_request.body = self.get_request_body(params)
+
+ return mock_request
+
+ def test_good_custom_params(self):
+ """
+ Custom parameters are presented in right format.
+ """
+ self.xblock.custom_parameters = ["test_custom_params=test_custom_param_value"]
+ self.xblock.get_client_key_secret = Mock(
+ return_value=("test_client_key", "test_client_secret")
+ )
+ self.xblock.oauth_params = Mock()
+ self.xblock.get_input_fields()
+ self.xblock.oauth_params.assert_called_with(
+ {"custom_test_custom_params": "test_custom_param_value"},
+ "test_client_key",
+ "test_client_secret",
+ )
+
+ def test_bad_custom_params(self):
+ """
+ Custom parameters are presented in wrong format.
+ """
+ bad_custom_params = ["test_custom_params: test_custom_param_value"]
+ self.xblock.custom_parameters = bad_custom_params
+ self.xblock.get_client_key_secret = Mock(
+ return_value=("test_client_key", "test_client_secret")
+ )
+ self.xblock.oauth_params = Mock()
+ with pytest.raises(LTIError):
+ self.xblock.get_input_fields()
+
+ def test_max_score(self):
+ self.xblock.weight = 100.0
+
+ assert not self.xblock.has_score
+ assert self.xblock.max_score() is None
+
+ self.xblock.has_score = True
+
+ assert self.xblock.max_score() == 100.0
+
+ def test_context_id(self):
+ """
+ Tests that LTI parameter context_id is equal to course_id.
+ """
+ assert str(self.course_id) == self.xblock.context_id