diff --git a/edx_exams/apps/lti/tests/test_views.py b/edx_exams/apps/lti/tests/test_views.py index cf6ca1ed..c5579d12 100644 --- a/edx_exams/apps/lti/tests/test_views.py +++ b/edx_exams/apps/lti/tests/test_views.py @@ -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, @@ -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/' @@ -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), @@ -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'), @@ -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): """ @@ -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( diff --git a/edx_exams/apps/lti/views.py b/edx_exams/apps/lti/views.py index dade1b92..4b333971 100644 --- a/edx_exams/apps/lti/views.py +++ b/edx_exams/apps/lti/views.py @@ -23,7 +23,7 @@ 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, @@ -31,7 +31,7 @@ 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 @@ -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 = [ @@ -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} ' @@ -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) @@ -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. ' @@ -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)