Files
edx-platform/xmodule/tests/test_lti_unit.py

567 lines
23 KiB
Python

"""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
from zoneinfo import ZoneInfo
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 webob.request import Request
from xblock.field_data import DictFieldData
from xblock.fields import ScopeIds, Timedelta
from common.djangoapps.xblock_django.constants import ATTR_KEY_ANONYMOUS_USER_ID
from xmodule import lti_block
from xmodule.tests.helpers import StubUserService
from . import get_test_system
from xmodule.lti_2_util import LTIError as BuiltInLTIError
from xblocks_contrib.lti.lti_2_util import LTIError as ExtractedLTIError
@override_settings(LMS_BASE="edx.org")
class _TestLTIBase(TestCase):
"""Logic tests for LTI block."""
__test__ = False
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.lti_class = lti_block.reset_class()
if settings.USE_EXTRACTED_LTI_BLOCK:
cls.LTIError = ExtractedLTIError
else:
cls.LTIError = BuiltInLTIError
def setUp(self):
super().setUp()
self.environ = {'wsgi.url_scheme': 'http', 'REQUEST_METHOD': 'POST'}
self.request_body_xml_template = textwrap.dedent("""
<?xml version = "1.0" encoding = "UTF-8"?>
<imsx_POXEnvelopeRequest xmlns = "{namespace}">
<imsx_POXHeader>
<imsx_POXRequestHeaderInfo>
<imsx_version>V1.0</imsx_version>
<imsx_messageIdentifier>{messageIdentifier}</imsx_messageIdentifier>
</imsx_POXRequestHeaderInfo>
</imsx_POXHeader>
<imsx_POXBody>
<{action}>
<resultRecord>
<sourcedGUID>
<sourcedId>{sourcedId}</sourcedId>
</sourcedGUID>
<result>
<resultScore>
<language>en-us</language>
<textString>{grade}</textString>
</resultScore>
</result>
</resultRecord>
</{action}>
</imsx_POXBody>
</imsx_POXEnvelopeRequest>
""")
self.course_id = CourseKey.from_string('org/course/run')
self.runtime = get_test_system(self.course_id)
self.runtime.publish = Mock()
self.runtime._services['rebind_user'] = Mock() # pylint: disable=protected-access
self.xblock = self.lti_class(
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)) # lint-amnesty, pylint: disable=line-too-long
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(
'xmodule.lti_block.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(
'xmodule.lti_block.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(ZoneInfo("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('xmodule.lti_block.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(self.LTIError):
self.xblock.get_client_key_secret()
@patch('xmodule.lti_block.signature.verify_hmac_sha1', Mock(return_value=True))
@patch(
'xmodule.lti_block.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('xmodule.lti_block.LTIBlock.get_outcome_service_url', Mock(return_value='https://testurl/'))
@patch('xmodule.lti_block.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'<?xml version=\'1.0\' encoding=\'utf-8\'?>\n'
b'<imsx_POXEnvelopeRequest xmlns="http://www.imsglobal.org/services/ltiv1p1/xsd/imsoms_v1p0">'
b'<imsx_POXHeader><imsx_POXRequestHeaderInfo><imsx_version>V1.0</imsx_version>'
b'<imsx_messageIdentifier>edX_fix</imsx_messageIdentifier></imsx_POXRequestHeaderInfo>'
b'</imsx_POXHeader><imsx_POXBody><replaceResultRequest><resultRecord><sourcedGUID>'
b'<sourcedId>MITxLTI/MITxLTI/201x:localhost%3A8000-i4x-MITxLTI-MITxLTI-lti-3751833a214a4f66a0d18f63234207f2'
b':363979ef768ca171b50f9d1bfb322131</sourcedId>'
b'</sourcedGUID><result><resultScore><language>en</language><textString>0.32</textString></resultScore>'
b'</result></resultRecord></replaceResultRequest></imsx_POXBody></imsx_POXEnvelopeRequest>'
)
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('xmodule.lti_block.signature.verify_hmac_sha1', Mock(return_value=False))
@patch(
'xmodule.lti_block.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(self.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(self.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
@override_settings(USE_EXTRACTED_LTI_BLOCK=True)
class TestLTIExtracted(_TestLTIBase):
__test__ = True
@override_settings(USE_EXTRACTED_LTI_BLOCK=False)
class TestLTIBuiltIn(_TestLTIBase):
__test__ = True