Merge pull request openedx#34021 from openedx/varshamenon4/refactor-c…

…ert-signals

refactor: move rejected exam event handlers

lms/djangoapps/certificates/api.py

@@ -929,9 +929,9 @@ def invalidate_certificate(user_id, course_key_or_id, source):
     Invalidate the user certificate in a given course if it exists and the user is not on the allowlist for this
     course run.
 
-    This function is called in services.py and handlers.py within the certificates folder. As of now,
+    This function is called in services.py and signals.py within the certificates folder. As of now,
     The call in services.py occurs when an exam attempt is rejected in the legacy exams backend, edx-proctoring.
-    The call in handlers.py is occurs when an exam attempt is rejected in the newer exams backend, edx-exams.
+    The call in signals.py is occurs when an exam attempt is rejected in the newer exams backend, edx-exams.
     """
     course_key = _get_key(course_key_or_id, CourseKey)
     if _is_on_certificate_allowlist(user_id, course_key):

lms/djangoapps/certificates/apps.py

@@ -23,7 +23,6 @@ def ready(self):
         # Can't import models at module level in AppConfigs, and models get
         # included from the signal handlers
         from lms.djangoapps.certificates import signals  # pylint: disable=unused-import
-        from lms.djangoapps.certificates import handlers  # pylint: disable=unused-import
         if settings.FEATURES.get('ENABLE_SPECIAL_EXAMS'):
             from lms.djangoapps.certificates.services import CertificateService
             set_runtime_service('certificates', CertificateService())

lms/djangoapps/certificates/signals.py

@@ -4,6 +4,7 @@
 
 import logging
 
+from django.contrib.auth import get_user_model
 from django.db.models.signals import post_save
 from django.dispatch import receiver
 
@@ -22,14 +23,20 @@
     CertificateStatuses,
     GeneratedCertificate
 )
-from lms.djangoapps.certificates.api import auto_certificate_generation_enabled
+from lms.djangoapps.certificates.api import (
+    auto_certificate_generation_enabled,
+    invalidate_certificate
+)
 from lms.djangoapps.verify_student.services import IDVerificationService
 from openedx.core.djangoapps.content.course_overviews.signals import COURSE_PACING_CHANGED
 from openedx.core.djangoapps.signals.signals import (
     COURSE_GRADE_NOW_FAILED,
     COURSE_GRADE_NOW_PASSED,
     LEARNER_NOW_VERIFIED
 )
+from openedx_events.learning.signals import EXAM_ATTEMPT_REJECTED
+
+User = get_user_model()
 
 log = logging.getLogger(__name__)
 
@@ -156,3 +163,17 @@ def _listen_for_enrollment_mode_change(sender, user, course_key, mode, **kwargs)
                 course_key,
             )
             return False
+
+
+@receiver(EXAM_ATTEMPT_REJECTED)
+def handle_exam_attempt_rejected_event(sender, signal, **kwargs):
+    """
+    Consume `EXAM_ATTEMPT_REJECTED` events from the event bus.
+    Pass the received data to invalidate_certificate in the services.py file in this folder.
+    """
+    event_data = kwargs.get('exam_attempt')
+    user_data = event_data.student_user
+    course_key = event_data.course_key
+
+    # Note that the course_key is the same as the course_key_or_id, and is being passed in as the course_key param
+    invalidate_certificate(user_data.id, course_key, source='exam_event')

lms/djangoapps/certificates/tests/test_signals.py

@@ -2,11 +2,17 @@
 Unit tests for enabling self-generated certificates for self-paced courses
 and disabling for instructor-paced courses.
 """
-
+from datetime import datetime, timezone
 from unittest import mock
+from uuid import uuid4
 
 import ddt
+from django.test import TestCase
 from edx_toggles.toggles.testutils import override_waffle_flag, override_waffle_switch
+from opaque_keys.edx.keys import CourseKey, UsageKey
+from openedx_events.data import EventsMetadata
+from openedx_events.learning.data import ExamAttemptData, UserData, UserPersonalData
+from openedx_events.learning.signals import EXAM_ATTEMPT_REJECTED
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory
 
@@ -18,6 +24,7 @@
     CertificateGenerationConfiguration,
     GeneratedCertificate
 )
+from lms.djangoapps.certificates.signals import handle_exam_attempt_rejected_event
 from lms.djangoapps.certificates.tests.factories import CertificateAllowlistFactory, GeneratedCertificateFactory
 from lms.djangoapps.grades.course_grade_factory import CourseGradeFactory
 from lms.djangoapps.grades.tests.utils import mock_passing_grade
@@ -433,3 +440,75 @@ def test_verified_to_audit(self):
         ) as mock_allowlist_task:
             self.verified_enrollment.change_mode('audit')
             mock_allowlist_task.assert_not_called()
+
+
+class ExamCompletionEventBusTests(TestCase):
+    """
+    Tests completion events from the event bus.
+    """
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.course_key = CourseKey.from_string('course-v1:edX+TestX+Test_Course')
+        cls.subsection_id = 'block-v1:edX+TestX+Test_Course+type@sequential+block@subsection'
+        cls.usage_key = UsageKey.from_string(cls.subsection_id)
+        cls.student_user = UserFactory(
+            username='student_user',
+        )
+
+    @staticmethod
+    def _get_exam_event_data(student_user, course_key, usage_key, exam_type, requesting_user=None):
+        """ create ExamAttemptData object for exam based event """
+        if requesting_user:
+            requesting_user_data = UserData(
+                id=requesting_user.id,
+                is_active=True,
+                pii=None
+            )
+        else:
+            requesting_user_data = None
+
+        return ExamAttemptData(
+            student_user=UserData(
+                id=student_user.id,
+                is_active=True,
+                pii=UserPersonalData(
+                    username=student_user.username,
+                    email=student_user.email,
+                ),
+            ),
+            course_key=course_key,
+            usage_key=usage_key,
+            requesting_user=requesting_user_data,
+            exam_type=exam_type,
+        )
+
+    @staticmethod
+    def _get_exam_event_metadata(event_signal):
+        """ create metadata object for event """
+        return EventsMetadata(
+            event_type=event_signal.event_type,
+            id=uuid4(),
+            minorversion=0,
+            source='openedx/lms/web',
+            sourcehost='lms.test',
+            time=datetime.now(timezone.utc)
+        )
+
+    @mock.patch('lms.djangoapps.certificates.signals.invalidate_certificate')
+    def test_exam_attempt_rejected_event(self, mock_api_function):
+        """
+        Assert that CertificateService api's invalidate_certificate is called upon consuming the event
+        """
+        exam_event_data = self._get_exam_event_data(self.student_user,
+                                                    self.course_key,
+                                                    self.usage_key,
+                                                    exam_type='proctored')
+        event_metadata = self._get_exam_event_metadata(EXAM_ATTEMPT_REJECTED)
+
+        event_kwargs = {
+            'exam_attempt': exam_event_data,
+            'metadata': event_metadata
+        }
+        handle_exam_attempt_rejected_event(None, EXAM_ATTEMPT_REJECTED, **event_kwargs)
+        mock_api_function.assert_called_once_with(self.student_user.id, self.course_key, source='exam_event')