feat: exam reset producer

edx_exams/apps/api/v1/views.py

@@ -26,6 +26,7 @@
 from edx_exams.apps.core.api import (
     check_if_exam_timed_out,
     create_exam_attempt,
+    delete_exam_attempt,
     get_active_attempt_for_user,
     get_attempt_by_id,
     get_course_exams,
@@ -626,7 +627,7 @@ def delete(self, request, attempt_id):
             error = {'detail': error_msg}
             return Response(status=status.HTTP_403_FORBIDDEN, data=error)
 
-        exam_attempt.delete()
+        delete_exam_attempt(exam_attempt, request.user)
         return Response(status=status.HTTP_204_NO_CONTENT)
 
 

edx_exams/apps/core/api.py

@@ -20,6 +20,7 @@
 from edx_exams.apps.core.signals.signals import (
     emit_exam_attempt_errored_event,
     emit_exam_attempt_rejected_event,
+    emit_exam_attempt_reset_event,
     emit_exam_attempt_submitted_event,
     emit_exam_attempt_verified_event
 )
@@ -142,6 +143,23 @@ def update_attempt_status(attempt_id, to_status):
     return attempt_id
 
 
+def delete_exam_attempt(attempt, requesting_user):
+    """
+    Delete or 'reset' an exam attempt
+    """
+    course_key = CourseKey.from_string(attempt.exam.course_id)
+    usage_key = UsageKey.from_string(attempt.exam.content_id)
+
+    attempt.delete()
+    emit_exam_attempt_reset_event(
+        attempt.user,
+        course_key,
+        usage_key,
+        attempt.exam.exam_type,
+        requesting_user
+    )
+
+
 def _allow_status_transition(attempt_obj, to_status):
     """
     Helper method to assert that a given status transition is allowed

edx_exams/apps/core/signals/handlers.py

@@ -6,6 +6,7 @@
 from openedx_events.learning.signals import (
     EXAM_ATTEMPT_ERRORED,
     EXAM_ATTEMPT_REJECTED,
+    EXAM_ATTEMPT_RESET,
     EXAM_ATTEMPT_SUBMITTED,
     EXAM_ATTEMPT_VERIFIED
 )
@@ -65,3 +66,17 @@ def listen_for_exam_attempt_errored(sender, signal, **kwargs):  # pylint: disabl
         event_data={'exam_attempt': kwargs['exam_attempt']},
         event_metadata=kwargs['metadata'],
     )
+
+
+@receiver(EXAM_ATTEMPT_RESET)
+def listen_for_exam_attempt_reset(sender, signal, **kwargs):  # pylint: disable=unused-argument
+    """
+    Publish EXAM_ATTEMPT_RESET signal onto the event bus
+    """
+    get_producer().send(
+        signal=EXAM_ATTEMPT_RESET,
+        topic='exam-attempt-reset',
+        event_key_field='exam_attempt.course_key',
+        event_data={'exam_attempt': kwargs['exam_attempt']},
+        event_metadata=kwargs['metadata'],
+    )

edx_exams/apps/core/signals/signals.py

@@ -6,6 +6,7 @@
 from openedx_events.learning.signals import (
     EXAM_ATTEMPT_ERRORED,
     EXAM_ATTEMPT_REJECTED,
+    EXAM_ATTEMPT_RESET,
     EXAM_ATTEMPT_SUBMITTED,
     EXAM_ATTEMPT_VERIFIED
 )
@@ -95,3 +96,22 @@ def emit_exam_attempt_errored_event(user, course_key, usage_key, exam_type):
             exam_type=exam_type
         )
     )
+
+
+def emit_exam_attempt_reset_event(user, course_key, usage_key, exam_type, requesting_user):
+    """
+    Emit the EXAM_ATTEMPT_RESET Open edX event.
+    """
+    user_data = _create_user_data(user)
+    requesting_user_data = _create_user_data(requesting_user)
+
+    # .. event_implemented_name: EXAM_ATTEMPT_RESET
+    EXAM_ATTEMPT_RESET.send_event(
+        exam_attempt=ExamAttemptData(
+            student_user=user_data,
+            course_key=course_key,
+            usage_key=usage_key,
+            exam_type=exam_type,
+            requesting_user=requesting_user_data
+        )
+    )

edx_exams/apps/core/tests/test_api.py

@@ -17,6 +17,7 @@
 from edx_exams.apps.core.api import (
     check_if_exam_timed_out,
     create_exam_attempt,
+    delete_exam_attempt,
     get_active_attempt_for_user,
     get_attempt_by_id,
     get_attempt_for_user_with_attempt_number_and_resource_id,
@@ -35,7 +36,7 @@
 )
 from edx_exams.apps.core.models import Exam, ExamAttempt
 from edx_exams.apps.core.statuses import ExamAttemptStatus
-from edx_exams.apps.core.test_utils.factories import ExamAttemptFactory, ExamFactory
+from edx_exams.apps.core.test_utils.factories import ExamAttemptFactory, ExamFactory, UserFactory
 
 test_start_time = datetime(2023, 11, 4, 11, 5, 23)
 test_time_limit_mins = 30
@@ -623,6 +624,62 @@ def test_exam_with_no_due_date(self):
         self.assertIsNotNone(ExamAttempt.objects.get(user_id=user_id, exam_id=exam_id))
 
 
+class TestDeleteExamAttempt(ExamsAPITestCase):
+    """
+    Tests for the API utility function `delete_exam_attempt`
+    """
+    def setUp(self):
+        super().setUp()
+
+        self.exam = ExamFactory()
+        self.student_user = UserFactory()
+        self.exam_attempt = ExamAttemptFactory(user=self.student_user, exam=self.exam)
+
+    def test_delete_exam_attempt(self):
+        """
+        Test that an exam attempt is deleted
+        """
+        delete_exam_attempt(self.exam_attempt, self.user)
+        self.assertFalse(ExamAttempt.objects.filter(id=self.exam_attempt.id).exists())
+
+    @patch('edx_exams.apps.core.signals.signals.EXAM_ATTEMPT_RESET.send_event')
+    def test_event_emitted(self, mock_event_send):
+        """
+        Test that when an exam attempt is deleted, the EXAM_ATTEMPT_RESET event is emitted.
+        """
+        delete_exam_attempt(self.exam_attempt, self.user)
+
+        user_data = UserData(
+            id=self.student_user.id,
+            is_active=self.student_user.is_active,
+            pii=UserPersonalData(
+                username=self.student_user.username,
+                email=self.student_user.email,
+                name=self.student_user.full_name
+            )
+        )
+        requesting_user_data = UserData(
+            id=self.user.id,
+            is_active=self.user.is_active,
+            pii=UserPersonalData(
+                username=self.user.username,
+                email=self.user.email,
+                name=self.user.full_name
+            )
+        )
+        course_key = CourseKey.from_string(self.exam.course_id)
+        usage_key = UsageKey.from_string(self.exam.content_id)
+
+        expected_data = ExamAttemptData(
+            student_user=user_data,
+            course_key=course_key,
+            usage_key=usage_key,
+            exam_type=self.exam.exam_type,
+            requesting_user=requesting_user_data,
+        )
+        mock_event_send.assert_called_once_with(exam_attempt=expected_data)
+
+
 class TestGetExamByContentId(ExamsAPITestCase):
     """
     Tests for the API utility function `get_exam_by_content_id`