feat: exam reset producer (#196)

send event to the event bus when an exam attempt is reset

edx_exams/apps/api/v1/views.py

@@ -36,6 +36,7 @@
     get_exam_by_id,
     get_provider_by_exam_id,
     is_exam_passed_due,
+    reset_exam_attempt,
     update_attempt_status
 )
 from edx_exams.apps.core.exam_types import get_exam_type
@@ -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()
+        reset_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,27 @@ def update_attempt_status(attempt_id, to_status):
     return attempt_id
 
 
+def reset_exam_attempt(attempt, requesting_user):
+    """
+    Reset an exam attempt
+    """
+    course_key = CourseKey.from_string(attempt.exam.course_id)
+    usage_key = UsageKey.from_string(attempt.exam.content_id)
+
+    log.info(
+        f'Resetting exam attempt for user_id={attempt.user.id} in exam={attempt.exam.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

@@ -7,6 +7,7 @@
 from openedx_events.learning.signals import (
     EXAM_ATTEMPT_ERRORED,
     EXAM_ATTEMPT_REJECTED,
+    EXAM_ATTEMPT_RESET,
     EXAM_ATTEMPT_SUBMITTED,
     EXAM_ATTEMPT_VERIFIED
 )
@@ -68,3 +69,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=topic_name,
+        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

@@ -25,6 +25,7 @@
     get_exam_by_content_id,
     get_exam_url_path,
     is_exam_passed_due,
+    reset_exam_attempt,
     update_attempt_status
 )
 from edx_exams.apps.core.exceptions import (
@@ -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 TestResetExamAttempt(ExamsAPITestCase):
+    """
+    Tests for the API utility function `reset_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_reset_exam_attempt(self):
+        """
+        Test that an exam attempt is deleted
+        """
+        reset_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 reset, the EXAM_ATTEMPT_RESET event is emitted.
+        """
+        reset_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`