From 0bf676c8306da5b638091bacf255d84d2a7f8812 Mon Sep 17 00:00:00 2001 From: Varsha Menon Date: Mon, 8 Jan 2024 15:10:35 -0500 Subject: [PATCH] refactor: move rejected exam event handlers --- lms/djangoapps/certificates/api.py | 4 +- lms/djangoapps/certificates/apps.py | 1 - lms/djangoapps/certificates/handlers.py | 28 ------ lms/djangoapps/certificates/signals.py | 23 ++++- .../certificates/tests/test_handlers.py | 87 ------------------- .../certificates/tests/test_signals.py | 81 ++++++++++++++++- 6 files changed, 104 insertions(+), 120 deletions(-) delete mode 100644 lms/djangoapps/certificates/handlers.py delete mode 100644 lms/djangoapps/certificates/tests/test_handlers.py diff --git a/lms/djangoapps/certificates/api.py b/lms/djangoapps/certificates/api.py index c93bc85a42f..b72113b9478 100644 --- a/lms/djangoapps/certificates/api.py +++ b/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): diff --git a/lms/djangoapps/certificates/apps.py b/lms/djangoapps/certificates/apps.py index f28bfa63a0b..50e5442c28b 100644 --- a/lms/djangoapps/certificates/apps.py +++ b/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()) diff --git a/lms/djangoapps/certificates/handlers.py b/lms/djangoapps/certificates/handlers.py deleted file mode 100644 index 8d45468497c..00000000000 --- a/lms/djangoapps/certificates/handlers.py +++ /dev/null @@ -1,28 +0,0 @@ -""" -Handlers for credits -""" -import logging - -from django.contrib.auth import get_user_model -from django.dispatch import receiver -from openedx_events.learning.signals import EXAM_ATTEMPT_REJECTED - -from lms.djangoapps.certificates.api import invalidate_certificate - -User = get_user_model() - -log = logging.getLogger(__name__) - - -@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') diff --git a/lms/djangoapps/certificates/signals.py b/lms/djangoapps/certificates/signals.py index 28e68e34025..d8db7bbf9ce 100644 --- a/lms/djangoapps/certificates/signals.py +++ b/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,7 +23,10 @@ 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 ( @@ -30,6 +34,9 @@ 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') diff --git a/lms/djangoapps/certificates/tests/test_handlers.py b/lms/djangoapps/certificates/tests/test_handlers.py deleted file mode 100644 index 241d9500bd6..00000000000 --- a/lms/djangoapps/certificates/tests/test_handlers.py +++ /dev/null @@ -1,87 +0,0 @@ -""" -Unit tests for certificates signals -""" -from datetime import datetime, timezone -from unittest import mock -from uuid import uuid4 - -from django.test import TestCase -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 common.djangoapps.student.tests.factories import UserFactory -from lms.djangoapps.certificates.handlers import handle_exam_attempt_rejected_event - - -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.handlers.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') diff --git a/lms/djangoapps/certificates/tests/test_signals.py b/lms/djangoapps/certificates/tests/test_signals.py index 320eec4d61d..d475cffbfb6 100644 --- a/lms/djangoapps/certificates/tests/test_signals.py +++ b/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')