Skip to content

Commit

Permalink
fix: lti bump
Browse files Browse the repository at this point in the history
  • Loading branch information
@ilee2u
ilee2u committed Jul 27, 2023
1 parent 2e27ca2 commit 68b5a1e
Show file tree
Hide file tree
Showing 2 changed files with 133 additions and 21 deletions.
105 changes: 103 additions & 2 deletions edx_exams/apps/lti/tests/test_views.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
from lti_consumer.lti_1p3.extensions.rest_framework.authentication import Lti1p3ApiAuthentication
from lti_consumer.models import LtiConfiguration, LtiProctoringConsumer

from edx_exams.apps.core.models import AssessmentControlResult
from edx_exams.apps.api.test_utils import ExamsAPITestCase, UserFactory
from edx_exams.apps.api.test_utils.factories import (
CourseExamConfigurationFactory,
Expand All @@ -38,7 +39,11 @@ def setUp(self):

self.course_exam_config = CourseExamConfigurationFactory()
self.exam = ExamFactory()

# Create an Exam Attempt that has already been submitted.
self.attempt = ExamAttemptFactory()
self.attempt.status = ExamAttemptStatus.submitted
self.attempt.save()

# Variables required for testing and verification
ISS = 'http://test-platform.example/'
Expand Down Expand Up @@ -129,6 +134,7 @@ def create_request_body(self, attempt_number, action, reason_code=None, incident
})
return request_body

# Test that an ACS result is created with the expected type
@ ddt.data(
(ExamAttemptStatus.ready_to_start, 200),
(ExamAttemptStatus.started, 200),
Expand Down Expand Up @@ -199,6 +205,99 @@ def test_acs_no_attempt_found(self,

self.assertEqual(response.status_code, 400)

@ ddt.data(
('user', ''),
('user', 'sub'),
('resource_link', ''),
('resource_link', 'id'),
('attempt_number', ''),
('action', ''),
)
@ ddt.unpack
@ patch.object(Lti1p3ApiAuthentication, 'authenticate', return_value=(AnonymousUser(), None))
@ patch('edx_exams.apps.lti.views.LtiProctoringAcsPermissions.has_permission')
@ patch('edx_exams.apps.lti.views.get_attempt_for_user_with_attempt_number_and_resource_id')
def test_acs_base_parameter_missing_errors(self,
acs_parameter,
acs_sub_parameter,
mock_get_attempt,
mock_permissions,
mock_authentication): # pylint: disable=unused-argument
"""
Test that the endpoint errors correctly if base ACS request parameters are not present
"""
# Make requests missing these items (create request but set the item passed in to undefined/None)
# Make the request
# Assert the correct error is thrown in the response with status 400
mock_get_attempt.return_value = self.attempt
mock_permissions.return_value = True

token = self.make_access_token('https://purl.imsglobal.org/spec/lti-ap/scope/control.all')

request_body = self.create_request_body(
self.attempt.attempt_number,
'terminate',
)

# Delete the selected field from the request body before sending to cause an error
if acs_sub_parameter == '':
del request_body[acs_parameter]
key_to_fail = acs_parameter
else:
del request_body[acs_parameter][acs_sub_parameter]
key_to_fail = acs_sub_parameter

# Even though the client.post function below uses json.dumps to serialize the request as json,
# The json serialization needs to happen before the request for an unknown reason
request_body = json.dumps(request_body)
response = self.client.post(self.url, data=request_body, content_type='application/json',
HTTP_AUTHORIZATION='Bearer {}'.format(token))
self.attempt.refresh_from_db()
self.assertEqual(response.data, f'ERROR: required parameter \'{key_to_fail}\' was not found.')

# TODO: Add tests for termination errors
@ ddt.data(
['reason_code'],
['incident_time'],
['incident_severity'],
)
@ ddt.unpack
@ patch.object(Lti1p3ApiAuthentication, 'authenticate', return_value=(AnonymousUser(), None))
@ patch('edx_exams.apps.lti.views.LtiProctoringAcsPermissions.has_permission')
@ patch('edx_exams.apps.lti.views.get_attempt_for_user_with_attempt_number_and_resource_id')
def test_acs_terminate_parameter_errors(self,
acs_parameter,
mock_get_attempt,
mock_permissions,
mock_authentication): # pylint: disable=unused-argument
"""
Test the endpoint errors correctly if request parameters required for the terminate action are not present
"""
# Make requests missing these items (create request but set the item passed in to undefined/None)
# Make the request
# Assert the correct error is thrown in the response with status 400
mock_get_attempt.return_value = self.attempt
mock_permissions.return_value = True

token = self.make_access_token('https://purl.imsglobal.org/spec/lti-ap/scope/control.all')

request_body = self.create_request_body(
self.attempt.attempt_number,
action='terminate',
reason_code='1',
incident_severity='0.1',
)

del request_body[acs_parameter]

# Even though the client.post function below uses json.dumps to serialize the request as json,
# The json serialization needs to happen before the request for an unknown reason
request_body = json.dumps(request_body)
response = self.client.post(self.url, data=request_body, content_type='application/json',
HTTP_AUTHORIZATION='Bearer {}'.format(token))
self.attempt.refresh_from_db()
self.assertEqual(response.data, f'ERROR: required parameter \'{acs_parameter}\' was not found.')

@ ddt.data(
# Testing reason codes with severity > 0.25
('0', '1.0', 'error'),
Expand Down Expand Up @@ -267,7 +366,9 @@ def test_acs_terminate(self,

self.assertEqual(self.attempt.status, expected_attempt_status)

# TODO: Implement error test for ACS termination
# Assure an entry was added to the ACResult model
data = AssessmentControlResult.objects.all()
self.assertEqual(len(data), 1)

def test_auth_failures(self):
"""
Expand Down Expand Up @@ -334,7 +435,7 @@ def test_start_proctoring_launch_data(self, mock_get_lti_launch_url):
attempt_number=self.attempt.attempt_number,
start_assessment_url=expected_proctoring_start_assessment_url,
assessment_control_url='http://test.exams:18740/lti/1/acs',
assessment_control_actions=['flagRequest'],
assessment_control_actions=['terminate'],
)

expected_launch_data = Lti1p3LaunchData(
Expand Down
49 changes: 30 additions & 19 deletions edx_exams/apps/lti/views.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,15 +23,15 @@
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response

from edx_exams.apps.api.constants import ASSEßSMENT_CONTROL_CODES
from edx_exams.apps.api.constants import ASSESSMENT_CONTROL_CODES
from edx_exams.apps.core.api import (
get_attempt_by_id,
get_attempt_for_user_with_attempt_number_and_resource_id,
get_exam_by_id,
update_attempt_status
)
from edx_exams.apps.core.exceptions import ExamIllegalStatusTransition
from edx_exams.apps.core.models import User
from edx_exams.apps.core.models import AssessmentControlResult, User
from edx_exams.apps.core.statuses import ExamAttemptStatus
from edx_exams.apps.lti.utils import get_lti_root

Expand Down Expand Up @@ -70,7 +70,8 @@ def acs(request, lti_config_id):
# ACS action to be performed
action = data['action']
except KeyError as err:
return Response(status=status.HTTP_400_BAD_REQUEST, data={'detail': err})
error_msg = f'ERROR: required parameter {err} was not found.'
return Response(status=status.HTTP_400_BAD_REQUEST, data=error_msg)

# Only flag ongoing or completed attempts
VALID_STATUSES = [
Expand Down Expand Up @@ -103,7 +104,7 @@ def acs(request, lti_config_id):

if action == 'flag':
# NOTE: The flag action is not yet supported.
# TODO: Make the flag action actually modify the exam attempt data (or perhaps another model?)
# If implemented, have it modify the exam attempt data (or perhaps another model?)
success_msg = (
f'NOTE: The flag action is not yet supported. The following is a placeholder message.'
f'Flagging exam attempt for user with id {anonymous_user_id} '
Expand All @@ -113,34 +114,31 @@ def acs(request, lti_config_id):
)
log.info(success_msg)

# NOTE: The code below is for the 'terminate' action, which is the only action we support currently.
# This code and its tests will need to be modified if other ACS actions are implemented.
elif action == 'terminate':
# Upon receiving a terminate request, the attempt referenced should have their status updated
# depending on the reason for termination (reason_code) and the incident_severity (scaling from 0.0 to 1.0).
# If the severity is greater than 0.25, then the attempt is marked for secondary review.

# Get the reason code for the termination and ensure it's a string to comply with the LTI specs
# See: http://www.imsglobal.org/spec/proctoring/v1p0#h.rsq8h6qxveab
# Get the termination paramenters
try:
reason_code = data['reason_code']
except KeyError:
error_msg = 'ERROR: required parameter reason_code was not found.'
error = {'detail': error_msg}
return Response(status=status.HTTP_400_BAD_REQUEST, data=error)
incident_time = data['incident_time']
severity = data['incident_severity']
except KeyError as err:
error_msg = f'ERROR: required parameter {err} was not found.'
return Response(status=status.HTTP_400_BAD_REQUEST, data=error_msg)

# Get the incident severity and ensure it's a string to comply with the LTI specs
severity = data['incident_severity']
# Ensure the incident_severity's a string to comply with the LTI specs
# See: http://www.imsglobal.org/spec/proctoring/v1p0#h.rsq8h6qxveab
if not isinstance(severity, str):
error_msg = 'ERROR: incident_severity must be passed to the ACS endpoint as a string per LTI specs.'
error = {'detail': error_msg}
return Response(status=status.HTTP_400_BAD_REQUEST, data=error)
return Response(status=status.HTTP_400_BAD_REQUEST, data=error_msg)

# Convert reason code and severity to correct formats
# NOTE: Is there a point in checking if severity is a string if we're just converting it into a
# float anyways?
severity = Decimal(severity)
SEVERITY_THRESHOLD = 0.25
reason_code_description = ASSESSMENT_CONTROL_CODES[reason_code]

# Regular submission occurred, but the learner did something
# that might be worth marking the attempt for review. Mark the attempt
# as requiring review based on the severity level (>0.25 -> needs review)
Expand Down Expand Up @@ -171,7 +169,6 @@ def acs(request, lti_config_id):
# be changed, or if we add another proctoring integration, then we may need to add a more
# precise elif condition here.
else:
# TODO: Decide if these are the codes we should count as errors
update_attempt_status(attempt.id, ExamAttemptStatus.error)
success_msg = (
f'Marked exam attempt as error. '
Expand All @@ -183,6 +180,20 @@ def acs(request, lti_config_id):
)
log.info(success_msg)

# Create a record of the ACS result
AssessmentControlResult.objects.create(
attempt=attempt,
action_type=action,
incident_time=incident_time,
severity=severity,
reason_code=reason_code,
)
log.info(
f'Created AssessmentControlResult for attempt with id {attempt.id}, '
f'action_type {action}, incident_time {incident_time}, severity {severity}, '
f'and reason_code {reason_code}.'
)

return Response(success_msg, status=200)


Expand Down

0 comments on commit 68b5a1e

Please sign in to comment.